# FourPoints
A production-grade Python library for real-time vehicle telemetry and diagnostics.
## Features
- **OBD-II Communication**: Connect to vehicle ECUs via USB or Bluetooth (including BLE)
- **Real-time Telemetry**: Stream vehicle data in real-time with asyncio support
- **Advanced Analytics**: Health diagnostics, predictive maintenance, and sensor status monitoring
- **AI-Powered Insights**: Gemini API integration for DTC explanations and maintenance advice
- **Location Tracking**: GPS integration for location data and trip statistics
- **Report Generation**: Create PDF/HTML reports with charts and vehicle data
- **FastAPI Endpoints**: RESTful API and WebSocket streaming for easy integration
- **Comprehensive Testing**: Unit tests with pytest and mocked OBD responses
## Installation
```bash
# Basic installation
pip install fourpoints
# With development tools
pip install fourpoints[dev]
# With documentation tools
pip install fourpoints[docs]
# Full installation with all dependencies
pip install fourpoints[all]
```
### Dependencies
FourPoints requires the `obd` package for OBD-II communication. This dependency is automatically installed with the basic installation.
### Gemini API Key
For AI-powered features (DTC explanations, maintenance advice, health insights), you'll need a Google Gemini API key:
1. Get your API key from [Google AI Studio](https://ai.google.dev/)
2. Set it as an environment variable:
```bash
export GEMINI_API_KEY="your_api_key_here"
```
Or on Windows:
```powershell
$env:GEMINI_API_KEY="your_api_key_here"
```
3. Alternatively, provide it directly when initializing the `GeminiClient` or in the `APIConfig`
## Quick Start
```python
from fourpoints.obd_client import OBDClient
from fourpoints.analytics import VehicleAnalytics
# Connect to vehicle
client = OBDClient()
client.connect()
# Get real-time telemetry
telemetry = client.get_telemetry()
print(f"RPM: {telemetry.get('RPM')}")
print(f"Speed: {telemetry.get('SPEED')} km/h")
# Check vehicle health
analytics = VehicleAnalytics(client)
health = analytics.get_health_status()
print(f"Health Score: {health['score']}/100")
```
## API Server
Start the API server to expose vehicle data via REST API and WebSocket:
```python
from fourpoints.api import FourPointsAPI, APIConfig
config = APIConfig(
host="localhost",
port=8000,
obd_port="/dev/ttyUSB0", # or auto-detect
enable_websocket=True,
gemini_api_key="YOUR_API_KEY" # For AI features
)
api = FourPointsAPI(config)
api.start()
```
## Real-time Streaming
Stream vehicle data in real-time with asyncio:
```python
import asyncio
from fourpoints.obd_client import OBDClient
from fourpoints.streaming import DataStream
async def main():
client = OBDClient()
client.connect()
stream = DataStream(client)
stream.add_commands(["RPM", "SPEED", "ENGINE_LOAD"])
# Set up event handlers
stream.on_data = lambda data: print(f"Data: {data}")
stream.on_threshold = lambda violations: print(f"Alert: {violations}")
# Set thresholds
stream.set_threshold("RPM", min_value=500, max_value=3000)
# Start streaming
stream.start()
# Run for 60 seconds
await asyncio.sleep(60)
# Stop streaming
stream.stop()
asyncio.run(main())
```
## WebSocket Streaming
Connect to the WebSocket endpoint to receive real-time data:
```javascript
// Browser JavaScript
const ws = new WebSocket('ws://localhost:8000/ws');
ws.onopen = () => {
// Subscribe to commands
ws.send(JSON.stringify({
action: 'subscribe',
commands: ['RPM', 'SPEED', 'ENGINE_LOAD']
}));
// Set threshold
ws.send(JSON.stringify({
action: 'set_threshold',
command: 'RPM',
min: 500,
max: 3000
}));
};
ws.onmessage = (event) => {
const message = JSON.parse(event.data);
if (message.type === 'data') {
console.log('Data:', message.data);
} else if (message.type === 'threshold') {
console.log('Threshold Alert:', message.violations);
}
};
```
## Example Script
Check out the [demo script](examples/demo.py) for a comprehensive example of using FourPoints.
## API Documentation
### OBD Client
```python
from fourpoints.obd_client import OBDClient, ConnectionType
# Create client
client = OBDClient(
port=None, # Auto-detect
connection_type=ConnectionType.AUTO, # AUTO, USB, BLUETOOTH, BLE, MOCK
timeout=2.0
)
# Connect to vehicle
client.connect()
# Get supported commands
commands = client.get_supported_commands()
# Query specific command
rpm = client.query_command("RPM")
# Get all telemetry
telemetry = client.get_telemetry()
# Get DTCs
dtcs = client.get_dtcs()
# Clear DTCs
client.clear_dtcs()
# Disconnect
client.disconnect()
```
### Analytics
```python
from fourpoints.analytics import VehicleAnalytics
# Create analytics
analytics = VehicleAnalytics(obd_client)
# Get health status
health = analytics.get_health_status()
# Get maintenance recommendations
maintenance = analytics.get_maintenance_recommendations()
# Get sensor status
sensors = analytics.get_sensor_status()
# Predict failures
predictions = analytics.predict_failures()
# Calculate health score
score = analytics.calculate_health_score()
```
### Gemini AI Integration
```python
from fourpoints.gemini_client import GeminiClient
# Create client with explicit API key
gemini = GeminiClient(api_key="YOUR_API_KEY")
# Or use environment variable (recommended)
# export GEMINI_API_KEY="your_api_key_here"
gemini = GeminiClient() # Will use GEMINI_API_KEY environment variable
# Get DTC explanation
explanation = gemini.explain_dtc("P0123")
# Get maintenance advice
advice = gemini.get_maintenance_advice(
dtcs=["P0123"],
telemetry={"RPM": 1500, "ENGINE_LOAD": 40}
)
# Get health insights
insights = gemini.get_health_insights(
telemetry={"RPM": 1500, "ENGINE_LOAD": 40}
)
```
### Location Tracking
```python
from fourpoints.location import LocationTracker, LocationSource
# Create tracker
tracker = LocationTracker()
# Start tracking
tracker.start_tracking(
source=LocationSource.GPSD, # GPSD, SERIAL, MOCK
port="/dev/ttyUSB1" # For SERIAL source
)
# Get current location
location = tracker.get_current_location()
print(f"Lat: {location.latitude}, Lon: {location.longitude}")
# Get trip data
trip = tracker.get_trip_data()
print(f"Distance: {trip.distance} km")
# Reset trip
tracker.reset_trip()
# Stop tracking
tracker.stop_tracking()
```
### Report Generation
```python
from fourpoints.reports import ReportGenerator, ReportFormat
# Create generator
generator = ReportGenerator(output_dir="reports")
# Generate report
report_path = generator.generate_report(
format=ReportFormat.PDF, # PDF, HTML
telemetry=telemetry,
dtcs=dtcs,
health_status=health,
maintenance=maintenance,
location=location.to_dict(),
trip_data=trip.__dict__,
vehicle_info={"make": "Toyota", "model": "Camry", "year": 2020}
)
# Get all reports
reports = generator.get_all_reports()
# Clean up old reports
generator.cleanup_old_reports(max_age_days=30)
```
## License
MIT License
Raw data
{
"_id": null,
"home_page": "https://github.com/amrittmishra/fourpoints",
"name": "fourpoints",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "obd, vehicle, telemetry, diagnostics, automotive",
"author": "FourPoints Team",
"author_email": "FourPoints Team <info@fourpoints.example.com>",
"download_url": "https://files.pythonhosted.org/packages/3f/4b/371efb714f4bff4d2d3b2740834e94880a81ef53b4a682d576665e8d7eb2/fourpoints-0.1.7.tar.gz",
"platform": null,
"description": "# FourPoints\n\nA production-grade Python library for real-time vehicle telemetry and diagnostics.\n\n## Features\n\n- **OBD-II Communication**: Connect to vehicle ECUs via USB or Bluetooth (including BLE)\n- **Real-time Telemetry**: Stream vehicle data in real-time with asyncio support\n- **Advanced Analytics**: Health diagnostics, predictive maintenance, and sensor status monitoring\n- **AI-Powered Insights**: Gemini API integration for DTC explanations and maintenance advice\n- **Location Tracking**: GPS integration for location data and trip statistics\n- **Report Generation**: Create PDF/HTML reports with charts and vehicle data\n- **FastAPI Endpoints**: RESTful API and WebSocket streaming for easy integration\n- **Comprehensive Testing**: Unit tests with pytest and mocked OBD responses\n\n## Installation\n\n```bash\n# Basic installation\npip install fourpoints\n\n# With development tools\npip install fourpoints[dev]\n\n# With documentation tools\npip install fourpoints[docs]\n\n# Full installation with all dependencies\npip install fourpoints[all]\n```\n\n### Dependencies\n\nFourPoints requires the `obd` package for OBD-II communication. This dependency is automatically installed with the basic installation.\n\n### Gemini API Key\n\nFor AI-powered features (DTC explanations, maintenance advice, health insights), you'll need a Google Gemini API key:\n\n1. Get your API key from [Google AI Studio](https://ai.google.dev/)\n2. Set it as an environment variable:\n ```bash\n export GEMINI_API_KEY=\"your_api_key_here\"\n ```\n \n Or on Windows:\n ```powershell\n $env:GEMINI_API_KEY=\"your_api_key_here\"\n ```\n3. Alternatively, provide it directly when initializing the `GeminiClient` or in the `APIConfig`\n\n## Quick Start\n\n```python\nfrom fourpoints.obd_client import OBDClient\nfrom fourpoints.analytics import VehicleAnalytics\n\n# Connect to vehicle\nclient = OBDClient()\nclient.connect()\n\n# Get real-time telemetry\ntelemetry = client.get_telemetry()\nprint(f\"RPM: {telemetry.get('RPM')}\")\nprint(f\"Speed: {telemetry.get('SPEED')} km/h\")\n\n# Check vehicle health\nanalytics = VehicleAnalytics(client)\nhealth = analytics.get_health_status()\nprint(f\"Health Score: {health['score']}/100\")\n```\n\n## API Server\n\nStart the API server to expose vehicle data via REST API and WebSocket:\n\n```python\nfrom fourpoints.api import FourPointsAPI, APIConfig\n\nconfig = APIConfig(\n host=\"localhost\",\n port=8000,\n obd_port=\"/dev/ttyUSB0\", # or auto-detect\n enable_websocket=True,\n gemini_api_key=\"YOUR_API_KEY\" # For AI features\n)\n\napi = FourPointsAPI(config)\napi.start()\n```\n\n## Real-time Streaming\n\nStream vehicle data in real-time with asyncio:\n\n```python\nimport asyncio\nfrom fourpoints.obd_client import OBDClient\nfrom fourpoints.streaming import DataStream\n\nasync def main():\n client = OBDClient()\n client.connect()\n \n stream = DataStream(client)\n stream.add_commands([\"RPM\", \"SPEED\", \"ENGINE_LOAD\"])\n \n # Set up event handlers\n stream.on_data = lambda data: print(f\"Data: {data}\")\n stream.on_threshold = lambda violations: print(f\"Alert: {violations}\")\n \n # Set thresholds\n stream.set_threshold(\"RPM\", min_value=500, max_value=3000)\n \n # Start streaming\n stream.start()\n \n # Run for 60 seconds\n await asyncio.sleep(60)\n \n # Stop streaming\n stream.stop()\n\nasyncio.run(main())\n```\n\n## WebSocket Streaming\n\nConnect to the WebSocket endpoint to receive real-time data:\n\n```javascript\n// Browser JavaScript\nconst ws = new WebSocket('ws://localhost:8000/ws');\n\nws.onopen = () => {\n // Subscribe to commands\n ws.send(JSON.stringify({\n action: 'subscribe',\n commands: ['RPM', 'SPEED', 'ENGINE_LOAD']\n }));\n \n // Set threshold\n ws.send(JSON.stringify({\n action: 'set_threshold',\n command: 'RPM',\n min: 500,\n max: 3000\n }));\n};\n\nws.onmessage = (event) => {\n const message = JSON.parse(event.data);\n \n if (message.type === 'data') {\n console.log('Data:', message.data);\n } else if (message.type === 'threshold') {\n console.log('Threshold Alert:', message.violations);\n }\n};\n```\n\n## Example Script\n\nCheck out the [demo script](examples/demo.py) for a comprehensive example of using FourPoints.\n\n## API Documentation\n\n### OBD Client\n\n```python\nfrom fourpoints.obd_client import OBDClient, ConnectionType\n\n# Create client\nclient = OBDClient(\n port=None, # Auto-detect\n connection_type=ConnectionType.AUTO, # AUTO, USB, BLUETOOTH, BLE, MOCK\n timeout=2.0\n)\n\n# Connect to vehicle\nclient.connect()\n\n# Get supported commands\ncommands = client.get_supported_commands()\n\n# Query specific command\nrpm = client.query_command(\"RPM\")\n\n# Get all telemetry\ntelemetry = client.get_telemetry()\n\n# Get DTCs\ndtcs = client.get_dtcs()\n\n# Clear DTCs\nclient.clear_dtcs()\n\n# Disconnect\nclient.disconnect()\n```\n\n### Analytics\n\n```python\nfrom fourpoints.analytics import VehicleAnalytics\n\n# Create analytics\nanalytics = VehicleAnalytics(obd_client)\n\n# Get health status\nhealth = analytics.get_health_status()\n\n# Get maintenance recommendations\nmaintenance = analytics.get_maintenance_recommendations()\n\n# Get sensor status\nsensors = analytics.get_sensor_status()\n\n# Predict failures\npredictions = analytics.predict_failures()\n\n# Calculate health score\nscore = analytics.calculate_health_score()\n```\n\n### Gemini AI Integration\n\n```python\nfrom fourpoints.gemini_client import GeminiClient\n\n# Create client with explicit API key\ngemini = GeminiClient(api_key=\"YOUR_API_KEY\")\n\n# Or use environment variable (recommended)\n# export GEMINI_API_KEY=\"your_api_key_here\"\ngemini = GeminiClient() # Will use GEMINI_API_KEY environment variable\n\n# Get DTC explanation\nexplanation = gemini.explain_dtc(\"P0123\")\n\n# Get maintenance advice\nadvice = gemini.get_maintenance_advice(\n dtcs=[\"P0123\"],\n telemetry={\"RPM\": 1500, \"ENGINE_LOAD\": 40}\n)\n\n# Get health insights\ninsights = gemini.get_health_insights(\n telemetry={\"RPM\": 1500, \"ENGINE_LOAD\": 40}\n)\n```\n\n### Location Tracking\n\n```python\nfrom fourpoints.location import LocationTracker, LocationSource\n\n# Create tracker\ntracker = LocationTracker()\n\n# Start tracking\ntracker.start_tracking(\n source=LocationSource.GPSD, # GPSD, SERIAL, MOCK\n port=\"/dev/ttyUSB1\" # For SERIAL source\n)\n\n# Get current location\nlocation = tracker.get_current_location()\nprint(f\"Lat: {location.latitude}, Lon: {location.longitude}\")\n\n# Get trip data\ntrip = tracker.get_trip_data()\nprint(f\"Distance: {trip.distance} km\")\n\n# Reset trip\ntracker.reset_trip()\n\n# Stop tracking\ntracker.stop_tracking()\n```\n\n### Report Generation\n\n```python\nfrom fourpoints.reports import ReportGenerator, ReportFormat\n\n# Create generator\ngenerator = ReportGenerator(output_dir=\"reports\")\n\n# Generate report\nreport_path = generator.generate_report(\n format=ReportFormat.PDF, # PDF, HTML\n telemetry=telemetry,\n dtcs=dtcs,\n health_status=health,\n maintenance=maintenance,\n location=location.to_dict(),\n trip_data=trip.__dict__,\n vehicle_info={\"make\": \"Toyota\", \"model\": \"Camry\", \"year\": 2020}\n)\n\n# Get all reports\nreports = generator.get_all_reports()\n\n# Clean up old reports\ngenerator.cleanup_old_reports(max_age_days=30)\n```\n\n## License\n\nMIT License\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A production-grade Python library for real-time vehicle telemetry and diagnostics",
"version": "0.1.7",
"project_urls": {
"Bug Tracker": "https://github.com/fourpoints/fourpoints/issues",
"Documentation": "https://fourpoints.readthedocs.io/",
"Homepage": "https://github.com/fourpoints/fourpoints"
},
"split_keywords": [
"obd",
" vehicle",
" telemetry",
" diagnostics",
" automotive"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "8bad590a9103b6ac5ec724aaf03fa73bc9587de2a970867822eebbe354f7830d",
"md5": "0f622e70f4eb3b8fed03c05ad8bce191",
"sha256": "3b6a58bdbcb1c2de6cd0acc4669f678163051418c0dc4b0119d44f7577b7ab0d"
},
"downloads": -1,
"filename": "fourpoints-0.1.7-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0f622e70f4eb3b8fed03c05ad8bce191",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 35453,
"upload_time": "2025-07-11T20:56:24",
"upload_time_iso_8601": "2025-07-11T20:56:24.508907Z",
"url": "https://files.pythonhosted.org/packages/8b/ad/590a9103b6ac5ec724aaf03fa73bc9587de2a970867822eebbe354f7830d/fourpoints-0.1.7-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "3f4b371efb714f4bff4d2d3b2740834e94880a81ef53b4a682d576665e8d7eb2",
"md5": "e97498a9d5214bebc371c8475c35e57e",
"sha256": "faa236585b9ff2ef555e5d53d8744fd95f5203e337dcbe5e5f194a216f9166ef"
},
"downloads": -1,
"filename": "fourpoints-0.1.7.tar.gz",
"has_sig": false,
"md5_digest": "e97498a9d5214bebc371c8475c35e57e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 48747,
"upload_time": "2025-07-11T20:56:26",
"upload_time_iso_8601": "2025-07-11T20:56:26.083437Z",
"url": "https://files.pythonhosted.org/packages/3f/4b/371efb714f4bff4d2d3b2740834e94880a81ef53b4a682d576665e8d7eb2/fourpoints-0.1.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-11 20:56:26",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "amrittmishra",
"github_project": "fourpoints",
"github_not_found": true,
"lcname": "fourpoints"
}
JSON
