smarttyre_api

  1"""Smart Tyre API Client
  2This module provides a client for interacting with the SmartTyre API.
  3"""
  4
  5import json
  6import secrets
  7import time
  8
  9import requests
 10
 11from sign_util import SignUtil
 12
 13
 14class SmartTyreAPI:
 15    """
 16    A class to interact with the Smart Tyre API.
 17    """
 18
 19    def __init__(self, base_url, client_id, client_secret, sign_key):
 20        """
 21        Initializes the SmartTyreAPI with the necessary credentials.
 22
 23        Args:
 24            base_url (str): The base URL of the Smart Tyre API.
 25            client_id (str): The client ID for authentication.
 26            client_secret (str): The client secret for authentication.
 27            sign_key (str): The signing key used to generate the signature.
 28        """
 29        self.base_url = base_url
 30        self.client_id = client_id
 31        self.client_secret = client_secret
 32        self.sign_key = sign_key
 33
 34    def _new_header(self, need_access_token=True):
 35        if need_access_token:
 36            access_token = self.get_access_token()
 37
 38            return {
 39                "clientId": self.client_id,
 40                "timestamp": str(int(time.time() * 1000)),
 41                "nonce": secrets.token_hex(16),
 42                "accessToken": access_token,
 43            }
 44
 45        return {
 46            "clientId": self.client_id,
 47            "timestamp": str(int(time.time() * 1000)),
 48            "nonce": secrets.token_hex(16),
 49        }
 50
 51    def _new_signature(self, headers, body, params, paths):
 52        return SignUtil.sign(
 53            headers=headers,
 54            body=body,
 55            params=params,
 56            paths=paths,
 57            sign_key=self.sign_key,
 58        )
 59
 60    def _new_get_request(self, endpoint, params):
 61        url = f"{self.base_url}{endpoint}"
 62        headers = self._new_header()
 63        headers["sign"] = self._new_signature(headers, "", params, [])
 64        headers["Content-Type"] = "application/json"
 65        headers["Accept"] = "application/json"
 66
 67        response = requests.get(url, headers=headers, params=params, timeout=20)
 68        if response.status_code == 200:
 69            return response.json().get("data")
 70        return None
 71
 72    def _new_post_request(
 73        self, endpoint, body, need_access_token=True, returns_data=True
 74    ):
 75        url = f"{self.base_url}{endpoint}"
 76        headers = self._new_header(need_access_token)
 77        headers["sign"] = self._new_signature(headers, body, {}, [])
 78        headers["Content-Type"] = "application/json"
 79        headers["Accept"] = "application/json"
 80        response = requests.post(url, headers=headers, data=body, timeout=20)
 81        if response.status_code == 200 and returns_data:
 82            return response.json().get("data")
 83        if response.status_code == 200:
 84            return response.json().get("msg")
 85        return None
 86
 87    def get_access_token(self):
 88        """
 89        Obtains an access token from the Smart Tyre API.
 90
 91        Args:
 92            base_url (str): The base URL of the Smart Tyre API.
 93            client_id (str): The client ID for authentication.
 94            client_secret (str): The client secret for authentication.
 95            sign_key (str): The signing key used to generate the signature.
 96
 97        Returns:
 98            The access token received from the API if available or None if the request fails.
 99        """
100        endpoint = "/smartyre/openapi/auth/oauth20/authorize"
101
102        body = {
103            "clientId": self.client_id,
104            "clientSecret": self.client_secret,
105            "grantType": "client_credentials",
106        }
107
108        body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
109
110        response = self._new_post_request(
111            endpoint=endpoint,
112            body=body_str,
113            need_access_token=False,
114        )
115
116        return response.get("accessToken") if response else None
117
118    # Vehicle Management
119
120    def add_vehicle(self, vehicle_info):
121        """
122        Creates a new vehicle in the Smart Tyre system.
123        Args:
124            vehicle_info (dict): The information of the vehicle to be created.
125
126                The dictionary should contain:
127
128                - isTractor (int): 0 for non-tractor, 1 for tractor, 2 for trailer
129                - licensePlateNumber (str): The license plate number of the vehicle
130                - emptyWeight (str): The empty weight of the vehicle
131                - fullWeight (str): The full weight of the vehicle
132                - axleTypeId (str): The ID of the axle type
133                - modelId (str): The ID of the vehicle model
134                - orgId (str): The ID of the organization
135                - tboxId (str): The ID of the TBox. Optional
136                - vehicleChassisNumber (str): The chassis number of the vehicle. Optional
137
138        Example:
139            ```python
140            new_vehicle = {
141                "isTractor": 0,
142                "licensePlateNumber": "ABC123",
143                "emptyWeight": "1000",
144                "fullWeight": "2000",
145                "axleTypeId": "2",
146                "modelId": "31",
147                "orgId": "218"
148            }
149            ```
150
151        Note: The vehicle_info dictionary must follow the API requirements.
152        Check the API documentation for optional fields.
153
154        Returns:
155            The response from the API if available or None if the request fails.
156        """
157        endpoint = "/smartyre/openapi/vehicle/insert"
158
159        body_str = json.dumps(vehicle_info, separators=(",", ":"), ensure_ascii=False)
160
161        return self._new_post_request(
162            endpoint=endpoint,
163            body=body_str,
164            returns_data=False,
165        )
166
167    def update_vehicle(self, vehicle_info):
168        """
169        Updates an existing vehicle in the Smart Tyre system.
170        Args:
171            vehicle_info (dict): The information of the vehicle to be updated.
172
173                The dictionary should contain:
174
175                - id (str): The ID of the vehicle to be updated
176                - isTractor (int): 0 for non-tractor, 1 for tractor, 2 for trailer
177                - licensePlateNumber (str): The license plate number of the vehicle
178                - emptyWeight (str): The empty weight of the vehicle
179                - fullWeight (str): The full weight of the vehicle
180                - axleTypeId (str): The ID of the axle type
181                - modelId (str): The ID of the vehicle model
182                - orgId (str): The ID of the organization
183                - tboxId (str): The ID of the TBox. Optional
184                - vehicleChassisNumber (str): The chassis number of the vehicle. Optional
185
186
187        Example:
188             ```python
189            updated_vehicle = {
190                "id": "7609",
191                "isTractor": 0,
192                "licensePlateNumber": "XYZ789",
193                "emptyWeight": "1200",
194                "fullWeight": "2200",
195                "axleTypeId": "2",
196                "modelId": "31",
197                "orgId": "218",
198            }
199            ```
200        Note: The vehicle_info dictionary must follow the API requirements.
201        Check the API documentation for optional fields.
202
203        Returns:
204            The response from the API if available or None if the request fails.
205        """
206        endpoint = "/smartyre/openapi/vehicle/update"
207
208        body_str = json.dumps(vehicle_info, separators=(",", ":"), ensure_ascii=False)
209
210        return self._new_post_request(
211            endpoint=endpoint,
212            body=body_str,
213            returns_data=False,
214        )
215
216    def get_vehicle_list(self, params=None):
217        """
218        Obtains the list of vehicles from the Smart Tyre API.
219        Args:
220            params (dict): Optional parameters for filtering the vehicle list.
221                Example: {"page": "1", "pageSize": "10"}
222        Note: The params dictionary must follow the API requirements.
223        Returns:
224            The list of vehicles if available or None if the request fails.
225        """
226        endpoint = "/smartyre/openapi/vehicle/list"
227
228        return self._new_get_request(endpoint, params=params)
229
230    def get_vehicle_info(self, vehicle_id):
231        """
232        Obtains detailed information about a specific vehicle.
233        Args:
234            vehicle_id (str): The ID of the vehicle.
235        Returns:
236            The detailed information about the vehicle if available or None if the request fails.
237        """
238        if not vehicle_id:
239            return None
240
241        endpoint = "/smartyre/openapi/vehicle/detail"
242
243        params = {
244            "vehicleId": [str(vehicle_id)],
245        }
246
247        return self._new_get_request(endpoint, params=params)
248
249    # Tire Management
250
251    def add_tire(self, tire_info):
252        """
253        Adds a new tire to the Smart Tyre system.
254        Args:
255            tire_info (dict): The information of the tire to be added.
256
257                The dictionary should contain:
258                - tyreCode (str): The identification code of the tire
259                - tyreBrandId (str): The ID of the tire brand
260                - tyreSizeId (str): The ID of the tire size
261                - tyrePattern (str): The pattern of the tire
262                - initialTreadDepth (str): The initial tread depth of the tire
263                - orgId (str): The ID of the organization. Optional
264                - sensorId (str): The ID of the sensor. Optional
265                - totalDistance (str): The total distance of the tire (km). Optional
266                - loadIndex (str): The load index of the tire. Optional
267                - speedLevel (str): The speed level of the tire. Optional
268                - newTreadDepth (str): The new tread depth of the tire. Optional
269
270        Example:
271            ```python
272            new_tire = {
273                "tyreCode": "ABC123",
274                "tyreBrandId": "1",
275                "tyreSizeId": "121",
276                "tyrePattern": "Pattern A",
277                "initialTreadDepth": "10"
278            }
279            ```
280
281        Note: The tire_info dictionary must follow the API requirements.
282        Check the API documentation for optional fields.
283        Returns:
284            The response from the API if available or None if the request fails.
285        """
286        endpoint = "/smartyre/openapi/tyre/insert"
287
288        body_str = json.dumps(tire_info, separators=(",", ":"), ensure_ascii=False)
289
290        return self._new_post_request(
291            endpoint=endpoint,
292            body=body_str,
293            returns_data=False,
294        )
295
296    def update_tire(self, tire_info):
297        """
298        Updates an existing tire in the Smart Tyre system.
299        Args:
300            tire_info (dict): The information of the tire to be updated.
301
302                The dictionary should contain:
303                - id (str): The ID of the tire to be updated
304                - tyreCode (str): The identification code of the tire, this cannot be changed
305                - tyreBrandId (str): The ID of the tire brand
306                - tyreSizeId (str): The ID of the tire size
307                - tyrePattern (str): The pattern of the tire
308                - initialTreadDepth (str): The initial tread depth of the tire
309                - orgId (str): The ID of the organization. Optional
310                - sensorId (str): The ID of the sensor. Optional
311                - totalDistance (str): The total distance of the tire (km). Optional
312                - loadIndex (str): The load index of the tire. Optional
313                - speedLevel (str): The speed level of the tire. Optional
314                - newTreadDepth (str): The new tread depth of the tire. Optional
315        Example:
316            ```python
317            updated_tire = {
318                "id": "47414",
319                "tyreCode": "ABC123",
320                "tyreBrandId": "8",
321                "tyreSizeId": "121",
322                "tyrePattern": "Pattern B",
323                "initialTreadDepth": "12"
324            }
325            ```
326        Note: The tire_info dictionary must follow the API requirements.
327        Check the API documentation for optional fields.
328
329        Returns:
330            The response from the API if available or None if the request fails.
331        """
332        endpoint = "/smartyre/openapi/tyre/update"
333
334        body_str = json.dumps(tire_info, separators=(",", ":"), ensure_ascii=False)
335        return self._new_post_request(
336            endpoint=endpoint,
337            body=body_str,
338            returns_data=False,
339        )
340
341    def get_tires_info_by_vehicle(self, vehicle_id):
342        """
343        Obtains tire information for a specific vehicle.
344        Args:
345            vehicle_id (str): The ID of the vehicle.
346        Returns:
347            The tire information if available or None if the request fails.
348        """
349
350        if not vehicle_id:
351            return None
352
353        endpoint = "/smartyre/openapi/vehicle/tyre/data"
354
355        body_str = json.dumps({"vehicleId": vehicle_id}, separators=(",", ":"))
356
357        return self._new_post_request(
358            endpoint=endpoint,
359            body=body_str,
360        )
361
362    def get_tire_list(self, params=None):
363        """
364        Obtains the list of tires from the Smart Tyre API.
365        Args:
366            params (dict): Optional parameters for filtering the tire list.
367                Example: {"page": "1", "pageSize": "10"}
368        Note: The params dictionary must follow the API requirements.
369        Returns:
370            The list of tires if available or None if the request fails.
371        """
372
373        endpoint = "/smartyre/openapi/tyre/list"
374
375        return self._new_get_request(endpoint, params=params)
376
377    def get_tire_info(self, tire_id):
378        """
379        Obtains detailed information about a specific tire.
380        Args:
381            tire_id (str): The ID of the tire.
382        Returns:
383            The detailed information about the tire if available or None if the request fails.
384        """
385        endpoint = "/smartyre/openapi/tyre/detail"
386        params = {
387            "id": [str(tire_id)],
388        }
389        return self._new_get_request(endpoint, params=params)
390
391    def bind_tire_to_vehicle(self, vehicle_id, tire_code, axle_index, wheel_index):
392        """
393        Binds a tire to a vehicle in the Smart Tyre system.
394        Args:
395            vehicle_id (str): The ID of the vehicle.
396            tire_code (str): The ID of the tire.
397            axle_index (int): The index of the axle.
398            wheel_index (int): The index of the wheel.
399        Returns:
400            The response from the API if available or None if the request fails.
401        """
402        endpoint = "/smartyre/openapi/vehicle/tyre/bind"
403
404        body = {
405            "vehicleId": vehicle_id,
406            "tyreCode": tire_code,
407            "axleIndex": axle_index,
408            "wheelIndex": wheel_index,
409        }
410
411        body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
412
413        return self._new_post_request(
414            endpoint=endpoint,
415            body=body_str,
416            returns_data=False,
417        )
418
419    def unbind_tire_from_vehicle(self, vehicle_id, tire_id):
420        """
421        Unbinds a tire from a vehicle in the Smart Tyre system.
422        Args:
423            vehicle_id (str): The ID of the vehicle.
424            tire_id (str): The ID of the tire.
425        Returns:
426            The response from the API if available or None if the request fails.
427        """
428
429        endpoint = "/smartyre/openapi/vehicle/tyre/unbind"
430        body = {
431            "vehicleId": vehicle_id,
432            "tyreCode": tire_id,
433        }
434
435        body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
436        return self._new_post_request(
437            endpoint=endpoint, body=body_str, returns_data=False
438        )
439
440    # Tbox Management
441
442    def add_tbox(self, tbox_info):
443        """
444        Adds a new TBox to the Smart Tyre system.
445        Args:
446            tbox_info (dict): The information of the TBox to be added.
447
448                The dictionary should contain:
449                - tboxCode (str): The identification code of the TBox.
450                    Must be 12 characters from 0-9 and A-F.
451                - version (str): The version of the TBox. Optional
452                - orgId (str): The ID of the organization. Optional
453                - ioTCardNumber (str): The IoT card number of the TBox. Optional
454                - carrier (str): The carrier of the TBox. Optional
455                - remark (str): Additional remarks about the TBox. Optional
456
457        Note: The tbox_info dictionary must follow the API requirements.
458        Check the API documentation for optional fields.
459        Returns:
460            The response from the API if available or None if the request fails.
461        """
462        endpoint = "/smartyre/openapi/tbox/insert"
463
464        body_str = json.dumps(tbox_info, separators=(",", ":"), ensure_ascii=False)
465
466        return self._new_post_request(
467            endpoint=endpoint,
468            body=body_str,
469            returns_data=False,
470        )
471
472    def update_tbox(self, tbox_info):
473        """
474        Updates an existing TBox in the Smart Tyre system.
475        Args:
476            tbox_info (dict): The information of the TBox to be updated.
477
478                The dictionary should contain:
479                - id (str): The ID of the TBox to be updated.
480                - tboxCode (str): The identification code of the Tbox to be updated.
481                    Cannot be changed.
482                - version (str): The version of the TBox. Optional
483                - orgId (str): The ID of the organization. Optional
484                - ioTCardNumber (str): The IoT card number of the TBox. Optional
485                - carrier (str): The carrier of the TBox. Optional
486                - remark (str): Additional remarks about the TBox. Optional
487
488        Note: The tbox_info dictionary must follow the API requirements.
489        Check the API documentation for optional fields.
490
491        Returns:
492            The response from the API if available or None if the request fails.
493        """
494        endpoint = "/smartyre/openapi/tbox/update"
495
496        body_str = json.dumps(tbox_info, separators=(",", ":"), ensure_ascii=False)
497
498        return self._new_post_request(
499            endpoint=endpoint,
500            body=body_str,
501            returns_data=False,
502        )
503
504    def get_tboxes_list(self, params=None):
505        """
506        Obtains the list of TBoxes from the Smart Tyre API.
507
508        Args:
509            params (dict): Optional parameters for filtering the TBox list.
510                Example: {"page": "1", "pageSize": "10"}
511        Note: The params dictionary must follow the API requirements.
512        Returns:
513            The list of TBoxes if available or None if the request fails.
514        """
515        endpoint = "/smartyre/openapi/tbox/list"
516
517        return self._new_get_request(endpoint, params=params)
518
519    def get_tbox_info(self, tbox_id):
520        """
521        Obtains information about a specific TBox.
522        Args:
523            tbox_id (str): The ID of the TBox.
524        Returns:
525            The detailed information about the TBox if available or None if the request fails.
526        """
527        endpoint = "/smartyre/openapi/tbox/detail"
528        params = {
529            "id": [str(tbox_id)],
530        }
531        return self._new_get_request(endpoint, params=params)
532
533    # Sensor Management
534
535    def add_sensor(self, sensor_info):
536        """
537        Adds a new sensor to the Smart Tyre system.
538        Args:
539            sensor_info (dict): The information of the sensor to be added.
540
541                The dictionary should contain:
542                - sensorCode (str): The identification code of the sensor.
543                    Must be 12 characters from 0-9 and A-F.
544                - version (str): The version of the sensor. Optional
545                - orgId (str): The ID of the organization. Optional
546                - remark (str): Additional remarks about the sensor. Optional
547
548        Note: The sensor_info dictionary must follow the API requirements.
549        Check the API documentation for optional fields.
550        Returns:
551            The response from the API if available or None if the request fails.
552        """
553        endpoint = "/smartyre/openapi/sensor/insert"
554
555        body_str = json.dumps(sensor_info, separators=(",", ":"), ensure_ascii=False)
556
557        return self._new_post_request(
558            endpoint=endpoint,
559            body=body_str,
560            returns_data=False,
561        )
562
563    def update_sensor(self, sensor_info):
564        """
565        Updates an existing sensor in the Smart Tyre system.
566        Args:
567            sensor_info (dict): The information of the sensor to be updated.
568
569                The dictionary should contain:
570                - id (str): The ID of the sensor to be updated.
571                - sensorCode (str): The identification code of the sensor.
572                    Must be 12 characters from 0-9 and A-F.
573                - version (str): The version of the sensor. Optional
574                - orgId (str): The ID of the organization. Optional
575                - remark (str): Additional remarks about the sensor. Optional
576
577        Note: The sensor_info dictionary must follow the API requirements.
578        Check the API documentation for optional fields.
579        Returns:
580            The response from the API if available or None if the request fails.
581        """
582        endpoint = "/smartyre/openapi/sensor/update"
583
584        body_str = json.dumps(sensor_info, separators=(",", ":"), ensure_ascii=False)
585
586        return self._new_post_request(
587            endpoint=endpoint,
588            body=body_str,
589            returns_data=False,
590        )
591
592    def get_sensor_list(self, params=None):
593        """
594        Obtains the list of sensors from the Smart Tyre API.
595        Args:
596            params (dict): Optional parameters for filtering the sensor list.
597                Example: {"page": "1", "pageSize": "10"}
598        Note: The params dictionary must follow the API requirements.
599        Returns:
600            The list of sensors if available or None if the request fails.
601        """
602        endpoint = "/smartyre/openapi/sensor/list"
603
604        return self._new_get_request(endpoint, params=params)
605
606    def get_sensor_info(self, sensor_id):
607        """
608        Obtains information about a specific sensor.
609        Args:
610            sensor_id (str): The ID of the sensor.
611        Returns:
612            The detailed information about the sensor if available or None if the request fails.
613        """
614        endpoint = "/smartyre/openapi/sensor/detail"
615        params = {
616            "id": [str(sensor_id)],
617        }
618        return self._new_get_request(endpoint, params=params)
619
620    def bind_sensor_to_tire(self, tire_code, vehicle_id, axle_index, wheel_index, sensor_code):
621        """
622        Binds a sensor to a tire in the Smart Tyre system.
623        Args:
624            tire_code (str): The ID of the tire.
625            vehicle_id (str): The ID of the vehicle.
626            axle_index (int): The index of the axle.
627            wheel_index (int): The index of the wheel.
628            sensor_code (str): The ID of the sensor.
629        Returns:
630            The response from the API if available or None if the request fails.
631        """
632        endpoint = "/smartyre/openapi/tyre/sensor/bind"
633
634        body = {
635            "tyreCode": tire_code,
636            "vehicleId": vehicle_id,
637            "axleIndex": axle_index,
638            "wheelIndex": wheel_index,
639            "sensorCode": sensor_code,
640        }
641
642        body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
643
644        return self._new_post_request(
645            endpoint=endpoint,
646            body=body_str,
647            returns_data=False,
648        )
649
650    def unbind_sensor_from_tire(self, tire_code, vehicle_id, axle_index, wheel_index, sensor_code):
651        """
652        Unbinds a sensor from a tire in the Smart Tyre system.
653        Args:
654            tire_code (str): The ID of the tire.
655            vehicle_id (str): The ID of the vehicle.
656            axle_index (int): The index of the axle.
657            wheel_index (int): The index of the wheel.
658            sensor_code (str): The ID of the sensor.
659        Returns:
660            The response from the API if available or None if the request fails.
661        """
662        endpoint = "/smartyre/openapi/tyre/sensor/unbind"
663
664        body = {
665            "tyreCode": tire_code,
666            "vehicleId": vehicle_id,
667            "axleIndex": axle_index,
668            "wheelIndex": wheel_index,
669            "sensorCode": sensor_code,
670        }
671
672        body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
673
674        return self._new_post_request(
675            endpoint=endpoint,
676            body=body_str,
677            returns_data=False,
678        )
679
680    # Reference Data Management
681    def get_tire_brands(self):
682        """
683        Obtains the list of tire brands from the Smart Tyre API.
684        Returns:
685            The list of tire brands if available or None if the request fails.
686        """
687        endpoint = "/smartyre/openapi/tyre/brand/all"
688
689        return self._new_get_request(endpoint, params={})
690
691    def get_tire_sizes(self):
692        """
693        Obtains the list of tire sizes from the Smart Tyre API.
694        Returns:
695            The list of tire sizes if available or None if the request fails.
696        """
697        endpoint = "/smartyre/openapi/tyre/size/all"
698
699        return self._new_get_request(endpoint, params={})
700
701    def get_vehicle_models(self):
702        """
703        Obtains the list of vehicle models from the Smart Tyre API.
704        Returns:
705            The list of vehicle models if available or None if the request fails.
706        """
707        endpoint = "/smartyre/openapi/vehicle/model/all"
708
709        return self._new_get_request(endpoint, params={})
710
711    def get_axle_types(self):
712        """
713        Obtains the list of axle types from the Smart Tyre API.
714        Returns:
715            The list of axle types if available or None if the request fails.
716        """
717        endpoint = "/smartyre/openapi/vehicle/axle/all"
718
719        return self._new_get_request(endpoint, params={})