Fork pykeydelivery for CarbonTracer client
First working version
This commit is contained in:
parent
aa3eb4dacf
commit
3d5ee20c2a
12 changed files with 162 additions and 112 deletions
|
@ -9,9 +9,8 @@ before_script:
|
||||||
- source venv/bin/activate
|
- source venv/bin/activate
|
||||||
- pip install -U pip
|
- pip install -U pip
|
||||||
- pip install .
|
- pip install .
|
||||||
- echo "[KeyDelivery]" > config.ini
|
- echo "[CarbonTracer]" > config.ini
|
||||||
- echo "key = ${API_KEY}" >> config.ini
|
- echo "key = ${API_KEY}" >> config.ini
|
||||||
- echo "secret = ${API_SECRET}" >> config.ini
|
|
||||||
|
|
||||||
test:
|
test:
|
||||||
stage: test
|
stage: test
|
||||||
|
|
2
LICENSE
2
LICENSE
|
@ -1,4 +1,4 @@
|
||||||
Copyright (c) 2023 Kumi Mitterer <pykeydelivery@kumi.email>
|
Copyright (c) 2023 Kumi Mitterer <pycarbontracer@kumi.email>
|
||||||
|
|
||||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||||
of this software and associated documentation files (the "Software"), to deal
|
of this software and associated documentation files (the "Software"), to deal
|
||||||
|
|
26
README.md
26
README.md
|
@ -1,31 +1,35 @@
|
||||||
# KeyDelivery API Python Client
|
# CarbonTracer API Python Client
|
||||||
|
|
||||||
This is a Python client for the KeyDelivery API. It is a wrapper around the [KeyDelivery](https://kd100.com/) API, which allows you to track your shipments.
|
This is a Python client for CarbonTracer. It is a wrapper around the [CarbonTracer](https://carbontracer.uni-graz.at/) REST API, which allows you to calculate CO2 equivalents for personal transport.
|
||||||
|
|
||||||
It is not fully featured yet, but it is a good starting point.
|
|
||||||
|
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install git+https://kumig.it/kumitterer/pykeydelivery
|
pip install git+https://kumig.it/kumitterer/pycarbontracer.git
|
||||||
```
|
```
|
||||||
|
|
||||||
## Usage
|
## Usage
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from keydelivery import KeyDelivery
|
from pycarbontracer import CarbonTracer
|
||||||
|
|
||||||
api = KeyDelivery("YOUR_API_KEY", "YOUR_API_SECRET")
|
# Create a new CarbonTracer instance
|
||||||
|
|
||||||
# Find carrier by shipment number
|
ct = CarbonTracer("YOUR_API_KEY")
|
||||||
|
|
||||||
carrier_options = api.detect_carrier("YOUR_SHIPMENT_NUMBER")
|
# Or use CarbonTracer.from_config() if you have a config.ini in the format of config.dist.ini
|
||||||
|
|
||||||
# Realtime tracking
|
# Calculate the CO2 equivalents for a train trip from Graz to Vienna
|
||||||
|
|
||||||
tracking = api.realtime("CARRIER_CODE", "YOUR_SHIPMENT_NUMBER")
|
result = ct.routing("train", "8010 Graz", "1010 Wien")
|
||||||
|
|
||||||
|
# Print the result
|
||||||
|
|
||||||
|
print(f"CO2 equivalents: {result["response"]["data"]["co2eq"]} {result["response"]["data"]["unitco2eq"]}")
|
||||||
```
|
```
|
||||||
|
|
||||||
|
The `CarbonTracer` class also has methods for the `location`, `address` and `co2only` endpoints, as documented in the [CarbonTracer API documentation](https://carbontracer.uni-graz.at/api-doc). The documentation also includes information about the input and output parameters, so make sure to check it out.
|
||||||
|
|
||||||
## License
|
## License
|
||||||
|
|
||||||
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
|
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
|
|
@ -1,3 +1,2 @@
|
||||||
[KeyDelivery]
|
[CarbonTracer]
|
||||||
key = api_key
|
key = api_key
|
||||||
secret = api_secret
|
|
||||||
|
|
|
@ -3,12 +3,12 @@ requires = ["hatchling"]
|
||||||
build-backend = "hatchling.build"
|
build-backend = "hatchling.build"
|
||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "pykeydelivery"
|
name = "pycarbontracer"
|
||||||
version = "0.9.0"
|
version = "0.9.0"
|
||||||
authors = [
|
authors = [
|
||||||
{ name="Kumi Mitterer", email="pykeydelivery@kumi.email" },
|
{ name="Kumi Mitterer", email="pycarbontracer@kumi.email" },
|
||||||
]
|
]
|
||||||
description = "Simple Python wrapper to fetch data from KeyDelivery (kd100.com)"
|
description = "Simple Python wrapper to fetch data from CarbonTracer (carbontracer.uni-graz.at)"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = { file="LICENSE" }
|
license = { file="LICENSE" }
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
|
@ -19,5 +19,5 @@ classifiers = [
|
||||||
]
|
]
|
||||||
|
|
||||||
[project.urls]
|
[project.urls]
|
||||||
"Homepage" = "https://kumig.it/kumitterer/pykeydelivery"
|
"Homepage" = "https://kumig.it/kumitterer/pycarbontracer"
|
||||||
"Bug Tracker" = "https://kumig.it/kumitterer/pykeydelivery/issues"
|
"Bug Tracker" = "https://kumig.it/kumitterer/pycarbontracer/issues"
|
2
src/pycarbontracer/classes/__init__.py
Normal file
2
src/pycarbontracer/classes/__init__.py
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
from .http import HTTPRequest
|
||||||
|
from .carbontracer import CarbonTracer
|
114
src/pycarbontracer/classes/carbontracer.py
Normal file
114
src/pycarbontracer/classes/carbontracer.py
Normal file
|
@ -0,0 +1,114 @@
|
||||||
|
from hashlib import md5
|
||||||
|
from configparser import ConfigParser
|
||||||
|
from urllib.parse import urljoin, quote
|
||||||
|
|
||||||
|
import json
|
||||||
|
|
||||||
|
from .http import HTTPRequest
|
||||||
|
|
||||||
|
|
||||||
|
class CarbonTracer:
|
||||||
|
BASE_URL = "https://api.carbontracer.uni-graz.at/"
|
||||||
|
|
||||||
|
def __init__(self, key: str, base_url: str = BASE_URL):
|
||||||
|
self.key = key
|
||||||
|
self.base_url = base_url
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_config(cls, config: ConfigParser | str, section: str = "CarbonTracer") -> "CarbonTracer":
|
||||||
|
if isinstance(config, str):
|
||||||
|
temp_config = ConfigParser()
|
||||||
|
temp_config.read(config)
|
||||||
|
config = temp_config
|
||||||
|
|
||||||
|
key = config.get(section, "key")
|
||||||
|
base_url = config.get(section, "base_url", fallback=cls.BASE_URL)
|
||||||
|
|
||||||
|
return cls(key, base_url)
|
||||||
|
|
||||||
|
def get_request(self, endpoint: str, message: dict) -> HTTPRequest:
|
||||||
|
url = self.base_url + endpoint
|
||||||
|
|
||||||
|
request = HTTPRequest(url)
|
||||||
|
|
||||||
|
return request
|
||||||
|
|
||||||
|
def location(self, location: str) -> dict:
|
||||||
|
"""Request a location from a string.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
location (str): The location to request.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The response from the server.
|
||||||
|
"""
|
||||||
|
|
||||||
|
request = HTTPRequest(urljoin(self.base_url, "/".join(["location", self.key, quote(location)])))
|
||||||
|
|
||||||
|
return request.execute()
|
||||||
|
|
||||||
|
def address(self, postalCode: str, city: str, street: str):
|
||||||
|
"""Request an (Austrian) address from a postal code, city and street.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
postalCode (str): The postal code.
|
||||||
|
city (str): The city.
|
||||||
|
street (str): The street.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The response from the server.
|
||||||
|
"""
|
||||||
|
|
||||||
|
request = HTTPRequest(urljoin(self.base_url, "/".join(["address", self.key, postalCode, city, street])))
|
||||||
|
|
||||||
|
return request.execute()
|
||||||
|
|
||||||
|
def routing(self, type: str, start: str, dest: str, waypoints: bool = False, bbox: bool = False, airports: bool = False):
|
||||||
|
"""Make a routing request.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
type (str): The type of routing to request, e.g. "car" or "train". See API documentation for more information.
|
||||||
|
start (str): The start location as a string starting with ZIP code or lat,lon coordinates.
|
||||||
|
dest (str): The destination location as a string starting with ZIP code or lat,lon coordinates.
|
||||||
|
waypoints (bool, optional): Whether to include waypoints for map display in the response. Defaults to False.
|
||||||
|
bbox (bool, optional): Whether to include the bounding box for map display in the response. Defaults to False.
|
||||||
|
airports (bool, optional): For flights, calculate the route from the nearest airports to start and dest. Defaults to False.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The response from the server.
|
||||||
|
"""
|
||||||
|
|
||||||
|
url = urljoin(self.base_url, "/".join(["routing", self.key, type, quote(start), quote(dest)]))
|
||||||
|
|
||||||
|
options = []
|
||||||
|
|
||||||
|
if waypoints:
|
||||||
|
options.append("waypoints")
|
||||||
|
|
||||||
|
if bbox:
|
||||||
|
options.append("bbox")
|
||||||
|
|
||||||
|
if airports:
|
||||||
|
options.append("airports")
|
||||||
|
|
||||||
|
if options:
|
||||||
|
url = url + "/" + "options=" + ",".join(options)
|
||||||
|
|
||||||
|
request = HTTPRequest(url)
|
||||||
|
|
||||||
|
return request.execute()
|
||||||
|
|
||||||
|
def co2only(self, type: str, distance_km: int):
|
||||||
|
"""Only calculate the CO2 emissions for a given distance for a given transport type.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
type (str): The type of transport to calculate the CO2 emissions for, e.g. "car" or "train". See API documentation for more information.
|
||||||
|
distance_km (int): The distance in kilometers.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The response from the server.
|
||||||
|
"""
|
||||||
|
|
||||||
|
request = HTTPRequest(urljoin(self.base_url, "/".join(["co2only", self.key, type, str(distance_km)])))
|
||||||
|
|
||||||
|
return request.execute()
|
|
@ -4,7 +4,7 @@ import json
|
||||||
|
|
||||||
|
|
||||||
class HTTPRequest(Request):
|
class HTTPRequest(Request):
|
||||||
USER_AGENT = "Mozilla/5.0 (compatible; PyKeyDelivery/dev; +https://kumig.it/kumitterer/pykeydelivery)"
|
USER_AGENT = "Mozilla/5.0 (compatible; PyCarbonTracer/dev; +https://kumig.it/kumitterer/pycarbontracer)"
|
||||||
|
|
||||||
def __init__(self, *args, **kwargs):
|
def __init__(self, *args, **kwargs):
|
||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
|
@ -15,7 +15,3 @@ class HTTPRequest(Request):
|
||||||
if load_json:
|
if load_json:
|
||||||
response = json.loads(response)
|
response = json.loads(response)
|
||||||
return response
|
return response
|
||||||
|
|
||||||
def add_json_payload(self, payload: dict):
|
|
||||||
self.add_header("Content-Type", "application/json")
|
|
||||||
self.data = json.dumps(payload).encode("utf-8")
|
|
|
@ -1,2 +0,0 @@
|
||||||
from .http import HTTPRequest
|
|
||||||
from .keydelivery import KeyDelivery
|
|
|
@ -1,61 +0,0 @@
|
||||||
from hashlib import md5
|
|
||||||
from configparser import ConfigParser
|
|
||||||
|
|
||||||
import json
|
|
||||||
|
|
||||||
from .http import HTTPRequest
|
|
||||||
|
|
||||||
|
|
||||||
class KeyDelivery:
|
|
||||||
BASE_URL = "https://www.kd100.com/api/v1/"
|
|
||||||
|
|
||||||
def __init__(self, key: str, secret: str, base_url: str = BASE_URL):
|
|
||||||
self.key = key
|
|
||||||
self.secret = secret
|
|
||||||
self.base_url = base_url
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_config(cls, config: ConfigParser | str, section: str = "KeyDelivery") -> "KeyDelivery":
|
|
||||||
if isinstance(config, str):
|
|
||||||
temp_config = ConfigParser()
|
|
||||||
temp_config.read(config)
|
|
||||||
config = temp_config
|
|
||||||
|
|
||||||
key = config.get(section, "key")
|
|
||||||
secret = config.get(section, "secret")
|
|
||||||
base_url = config.get(section, "base_url", fallback=cls.BASE_URL)
|
|
||||||
|
|
||||||
return cls(key, secret, base_url)
|
|
||||||
|
|
||||||
def get_signature(self, message: dict) -> str:
|
|
||||||
content = json.dumps(message)
|
|
||||||
data = (content + self.key + self.secret).encode("utf-8")
|
|
||||||
return md5(data).hexdigest().upper()
|
|
||||||
|
|
||||||
def get_request(self, endpoint: str, message: dict) -> HTTPRequest:
|
|
||||||
url = self.base_url + endpoint
|
|
||||||
signature = self.get_signature(message)
|
|
||||||
|
|
||||||
request = HTTPRequest(url)
|
|
||||||
request.add_json_payload(message)
|
|
||||||
request.add_header("API-Key", self.key)
|
|
||||||
request.add_header("signature", signature)
|
|
||||||
|
|
||||||
return request
|
|
||||||
|
|
||||||
def realtime(self, carrier: str, tracking_number: str) -> bytes:
|
|
||||||
message = {
|
|
||||||
"carrier_id": carrier,
|
|
||||||
"tracking_number": tracking_number,
|
|
||||||
}
|
|
||||||
|
|
||||||
request = self.get_request("tracking/realtime", message)
|
|
||||||
return request.execute()
|
|
||||||
|
|
||||||
def detect_carrier(self, tracking_number: str) -> bytes:
|
|
||||||
message = {
|
|
||||||
"tracking_number": tracking_number,
|
|
||||||
}
|
|
||||||
|
|
||||||
request = self.get_request("carriers/detect", message)
|
|
||||||
return request.execute()
|
|
41
test.py
41
test.py
|
@ -3,7 +3,7 @@ from configparser import ConfigParser
|
||||||
|
|
||||||
import json
|
import json
|
||||||
|
|
||||||
from pykeydelivery import *
|
from pycarbontracer import *
|
||||||
|
|
||||||
class TestHTTPRequest(TestCase):
|
class TestHTTPRequest(TestCase):
|
||||||
def test_http_request(self):
|
def test_http_request(self):
|
||||||
|
@ -11,28 +11,27 @@ class TestHTTPRequest(TestCase):
|
||||||
response = http.execute()
|
response = http.execute()
|
||||||
self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT)
|
self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT)
|
||||||
|
|
||||||
def test_http_request_with_json_payload(self):
|
class TestCarbonTracer(TestCase):
|
||||||
http = HTTPRequest("https://httpbin.org/post")
|
def setUp(self):
|
||||||
http.add_json_payload({"foo": "bar"})
|
|
||||||
response = http.execute()
|
|
||||||
self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT)
|
|
||||||
self.assertEqual(response["headers"]["Content-Type"], "application/json")
|
|
||||||
self.assertEqual(response["json"]["foo"], "bar")
|
|
||||||
|
|
||||||
class TestKeyDelivery(TestCase):
|
|
||||||
def __init__(self, *args, **kwargs):
|
|
||||||
super().__init__(*args, **kwargs)
|
|
||||||
self.config = ConfigParser()
|
self.config = ConfigParser()
|
||||||
self.config.read("config.ini")
|
self.config.read("config.ini")
|
||||||
self.keydelivery = KeyDelivery.from_config(self.config)
|
self.carbontracer = CarbonTracer.from_config(self.config)
|
||||||
|
|
||||||
def test_detect_carrier(self):
|
def test_carbontracer_location(self):
|
||||||
response = self.keydelivery.detect_carrier("483432314669")
|
response = self.carbontracer.location("Graz")
|
||||||
self.assertEqual(response["code"], 200)
|
self.assertEqual(response["response"]["data"]["country"], "AT")
|
||||||
|
|
||||||
def test_realtime(self):
|
def test_carbontracer_address(self):
|
||||||
response = self.keydelivery.realtime("gls", "483432314669")
|
response = self.carbontracer.address("8010", "Graz", "Gartengasse")
|
||||||
self.assertEqual(response["code"], 200)
|
self.assertEqual(response["response"]["data"]["country"], "Austria")
|
||||||
|
|
||||||
if __name__ == "__main__":
|
def test_carbontracer_routing(self):
|
||||||
main()
|
response = self.carbontracer.routing("car", "8010 Graz", "1010 Wien", waypoints=True)
|
||||||
|
self.assertTrue("wayPoints" in response["response"]["data"])
|
||||||
|
self.assertEqual(response["response"]["data"]["requestType"], "car")
|
||||||
|
self.assertEqual(response["response"]["data"]["startLocation"]["postalCode"], "8010")
|
||||||
|
|
||||||
|
def test_carbontracer_co2only(self):
|
||||||
|
response = self.carbontracer.co2only("car", 100)
|
||||||
|
self.assertEqual(response["response"]["data"]["distance"], 100)
|
||||||
|
self.assertEqual(response["response"]["data"]["requestType"], "car")
|
Loading…
Reference in a new issue