128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
import logging
|
|
import sys
|
|
import json
|
|
import time
|
|
|
|
from src import VERSION, Client, OrderField
|
|
from src.exceptions import WhereIsException, UnauthorizedException
|
|
|
|
|
|
stdout_handler = logging.StreamHandler(stream=sys.stdout)
|
|
logging.basicConfig(
|
|
format="[%(levelname)s] - %(asctime)s - %(message)s",
|
|
level=logging.INFO,
|
|
handlers=(stdout_handler,),
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
"""
|
|
This a sample script to demonstrate how to deal with WhereIs API client.
|
|
|
|
All you need is to provide some environments variables describe in the `.env.example`.
|
|
You can:
|
|
- Copy `.env.example` into `.env` and feed env variables or,
|
|
- Directly export those mandatories variables
|
|
|
|
Once its done, you can initialize the Client:
|
|
|
|
cli = Client.from_env()
|
|
|
|
and use all the available methods to interact with the WhereIs API.
|
|
|
|
Sessions:
|
|
- create_session
|
|
- get_sessions
|
|
- get_session
|
|
- update_session
|
|
- delete_session
|
|
- update_session_users
|
|
- close_session
|
|
- watch_session_events
|
|
- stop_watch_session
|
|
WS token:
|
|
- get_wstokens
|
|
- create_wstoken
|
|
- delete_wstoken
|
|
GPS positions:
|
|
- get_gps_positions
|
|
"""
|
|
logging.info(f"WhereIs client v{VERSION}")
|
|
|
|
# initialize the client
|
|
try:
|
|
cli = Client.from_env()
|
|
except WhereIsException as e:
|
|
logging.error(f"unable to initialize WhereIs API client: {e}")
|
|
except Exception as e:
|
|
logging.error("unexpected error occurred while initializing client", exc_info=True)
|
|
|
|
# get ws tokens
|
|
tokens = cli.get_wstokens()
|
|
print(json.dumps(tokens, indent=2))
|
|
|
|
# delete ws token
|
|
if len(tokens) > 0:
|
|
token = cli.delete_wstoken(tokens[0].get("id"))
|
|
print(json.dumps(tokens, indent=2))
|
|
|
|
# create ws token
|
|
try:
|
|
token = cli.create_wstoken()
|
|
except WhereIsException as e:
|
|
logging.error(f"error occurred while creating a ws token, status code: {e.error_code}")
|
|
print(json.dumps(e.content, indent=2))
|
|
|
|
# retrieve all user/public sessions
|
|
sessions = cli.get_sessions()
|
|
print(json.dumps(sessions, indent=2))
|
|
|
|
# create and update a session (must have `Streamer` role)
|
|
try:
|
|
session = cli.create_session(name="RUN-01")
|
|
session = cli.update_session(session.get("id"), name="RUN-02", description="My second run")
|
|
except WhereIsException as e:
|
|
logging.error(f"error occurred while creating/updating a session, status code: {e.error_code}")
|
|
print(json.dumps(e.content, indent=2))
|
|
|
|
# retrieve all user/public sessions with ordering
|
|
sessions = cli.get_sessions(ordering=OrderField.AscDateStart)
|
|
print(json.dumps(sessions, indent=2))
|
|
|
|
# retrieve all user/public sessions with search
|
|
sessions = cli.get_sessions(search="run", ordering=OrderField.AscDateStart)
|
|
print(json.dumps(sessions, indent=2))
|
|
|
|
# retrieve session by id
|
|
session = cli.get_session(session.get("id"))
|
|
print(json.dumps(session, indent=2))
|
|
|
|
# update session users
|
|
try:
|
|
session = cli.update_session_users(session.get("id"), [])
|
|
except WhereIsException as e:
|
|
logging.error(f"error occurred while updating users session, status code: {e.error_code}")
|
|
print(json.dumps(e.content, indent=2))
|
|
|
|
# close a session
|
|
try:
|
|
session = cli.close_session("does-not-exist")
|
|
except WhereIsException as e:
|
|
logging.error(f"error occurred while closing session, status code: {e.error_code}")
|
|
print(json.dumps(e.content, indent=2))
|
|
|
|
# get session events
|
|
cli.watch_session_events(session.get("id"))
|
|
|
|
# get gps positions from "2024-12-25T23:00:00" to infinity...
|
|
gps_positions = cli.get_gps_positions(date_start="2024-12-25T23:00:00")
|
|
print(json.dumps(gps_positions, indent=2))
|
|
|
|
# close the session
|
|
session = cli.close_session(session.get("id"))
|
|
|
|
# delete a session
|
|
cli.delete_session(session.get("id"))
|
|
|
|
# stop session events watcher
|
|
cli.stop_watch_session(session.get("id"), force=True)
|