getMySkodaData.py

import asyncio
import sys
import orjson
import contextlib
import io
from aiohttp import ClientSession, ClientTimeout
from aiohttp.client_exceptions import ClientResponseError
from myskoda import MySkoda
from myskoda.models.info import Info
from myskoda.models.health import Health
from myskoda.models.driving_range import DrivingRange

# Basic info to show (use info or vehicle): 
# - Type of car
# - Vehicle status
# - Driving Range
# - Vehicle image
#
# Check:
# - If VehicleStatus = VehicleStatus.ACTIVATED
#
# More advanced info:
# - Position
# - Service events
# - Status - doors
# - Trip Statistics
async def main(email, password, vin):
    session = None
    myskoda = None
    
    # Entering try block to avoid not closing session when any exception happens
    try:
        session = ClientSession(timeout=ClientTimeout(total=60))
        myskoda = MySkoda(session, mqtt_enabled=False)
        await myskoda.connect(email, password)

        # Get all needed data
        info: Info = await myskoda.get_info(vin)
        health: Health = await myskoda.get_health(vin)
        driving_range: DrivingRange = await myskoda.get_driving_range(vin)

        #Decode nested objects into JSON
        infoJSON = orjson.dumps(info).decode()
        healthJSON = orjson.dumps(health).decode()
        drivingRangeJSON = orjson.dumps(driving_range).decode()

        #Deserialize to Python dictionaries
        infoDict = orjson.loads(infoJSON)
        healthDict = orjson.loads(healthJSON)
        drivingRangeDict = orjson.loads(drivingRangeJSON)

        #Merge the dictionaries
        mySkodaDict = {**infoDict, **healthDict, **drivingRangeDict}

        # Fetch additional endpoints with error handling
        # Status endpoint
        try:
            with contextlib.redirect_stderr(io.StringIO()):
                status = await myskoda.get_status(vin)
                statusJSON = orjson.dumps(status).decode()
                statusDict = orjson.loads(statusJSON)
                mySkodaDict.update(statusDict)
        except (ClientResponseError, Exception):
            # Status endpoint may fail for some vehicles, continue without it
            pass

        # Positions endpoint
        try:
            with contextlib.redirect_stderr(io.StringIO()):
                positions = await myskoda.get_positions(vin)
                positionsJSON = orjson.dumps(positions).decode()
                positionsDict = orjson.loads(positionsJSON)
                mySkodaDict.update(positionsDict)
        except (ClientResponseError, Exception):
            # Positions endpoint may fail, continue without it
            pass

        # Maintenance endpoint
        try:
            with contextlib.redirect_stderr(io.StringIO()):
                maintenance = await myskoda.get_maintenance(vin)
                maintenanceJSON = orjson.dumps(maintenance).decode()
                maintenanceDict = orjson.loads(maintenanceJSON)
                mySkodaDict.update(maintenanceDict)
        except (ClientResponseError, Exception):
            # Maintenance endpoint may fail, continue without it
            pass

        # Charging endpoint (EV only)
        # Check if vehicle is electric before attempting to fetch charging data
        is_electric = (
            infoDict.get('specification', {}).get('engine', {}).get('type') == 'iV' or
            drivingRangeDict.get('car_type') == 'electric' or
            drivingRangeDict.get('primary_engine_range', {}).get('engine_type') == 'electric'
        )
        
        if is_electric:
            try:
                # Suppress stderr for this call to avoid noisy error logs for expected failures
                with contextlib.redirect_stderr(io.StringIO()):
                    charging = await myskoda.get_charging(vin)
                    chargingJSON = orjson.dumps(charging).decode()
                    chargingDict = orjson.loads(chargingJSON)
                    mySkodaDict.update(chargingDict)
            except (ClientResponseError, Exception) as e:
                # Charging endpoint may fail even for EVs (500 errors, etc.), continue without it
                # Suppress the error as it's expected in some cases
                pass

        #Serialize the merged dictionary back to a JSON string
        mySkodaJSON = orjson.dumps(mySkodaDict).decode()
    except Exception as e:
        # If any critical error occurs, log it and return error info
        error_msg = str(e)
        error_type = type(e).__name__
        print(f"ERROR: {error_type}: {error_msg}", file=sys.stderr)
        # Return error info as JSON so node_helper can handle it
        error_dict = {"error": f"{error_type}: {error_msg}"}
        mySkodaJSON = orjson.dumps(error_dict).decode()
    finally:
        #Closing connection
        try:
            if myskoda:
                await myskoda.disconnect()
            if session:
                await session.close()
        except:
            pass

        #Data returning to JS
        print(mySkodaJSON)
        sys.stdout.flush()

#Invoking the main function in CLI style
asyncio.run(main(sys.argv[1], sys.argv[2], sys.argv[3]))