Logging

logging

Flight logging module for Drone API.

Provides database logging for flight sessions, telemetry, and commands. Supports both SQLite (sync) and PostgreSQL (async) backends.

Example usage:

# SQLite (sync)
from pypack.logging import SQLiteLogger
logger = SQLiteLogger("flights.db")
session_id = logger.start_session(drone_id=1, notes="Test flight")
logger.log_telemetry(session_id, flight_data)
logger.end_session(session_id)

# PostgreSQL (async)
from pypack.logging import PostgresLogger
logger = PostgresLogger("postgresql://user:pass@localhost/drone")
await logger.connect()
session_id = await logger.start_session(drone_id=1)
await logger.log_telemetry(session_id, flight_data)
await logger.end_session(session_id)

FlightLogger

Bases: ABC

Abstract interface for flight logging backends.

Implementations must provide both sync and async versions of methods. The sync versions can be simple wrappers around async versions using asyncio.run() for backends that are natively async.

start_session abstractmethod

start_session(drone_id: int, notes: str | None = None) -> str

Start a new flight session.

Parameters:

Name	Type	Description	Default
drone_id	int	Drone identifier	required
notes	str | None	Optional session notes	None

Returns:

Type	Description
str	Session ID (UUID string)

end_session abstractmethod

end_session(session_id: str) -> None

End a flight session.

Parameters:

Name	Type	Description	Default
session_id	str	Session to end	required

get_session abstractmethod

get_session(session_id: str) -> FlightSession | None

Get session metadata.

Parameters:

Name	Type	Description	Default
session_id	str	Session ID	required

Returns:

Type	Description
FlightSession | None	FlightSession or None if not found

list_sessions abstractmethod

list_sessions(drone_id: int | None = None, start_after: datetime | None = None, limit: int = 100) -> list[FlightSession]

List flight sessions.

Parameters:

Name	Type	Description	Default
drone_id	int | None	Filter by drone ID	None
start_after	datetime | None	Filter sessions after this time	None
limit	int	Maximum sessions to return	100

Returns:

Type	Description
list[FlightSession]	List of FlightSession

log_telemetry abstractmethod

log_telemetry(session_id: str, data: FlightData) -> None

Log a telemetry data point.

Parameters:

Name	Type	Description	Default
session_id	str	Session ID	required
data	FlightData	Flight data to log	required

log_telemetry_batch abstractmethod

log_telemetry_batch(session_id: str, data: list[FlightData]) -> None

Log multiple telemetry data points in batch.

Parameters:

Name	Type	Description	Default
session_id	str	Session ID	required
data	list[FlightData]	List of flight data to log	required

get_telemetry abstractmethod

get_telemetry(session_id: str, start: datetime | None = None, end: datetime | None = None, limit: int = 10000) -> list[TelemetryRecord]

Retrieve telemetry for a session.

Parameters:

Name	Type	Description	Default
session_id	str	Session ID	required
start	datetime | None	Start time filter	None
end	datetime | None	End time filter	None
limit	int	Maximum records to return	10000

Returns:

Type	Description
list[TelemetryRecord]	List of TelemetryRecord

log_command abstractmethod

log_command(session_id: str, command: str, params: dict, result: int, duration_ms: float) -> None

Log a command execution.

Parameters:

Name	Type	Description	Default
session_id	str	Session ID	required
command	str	Command name	required
params	dict	Command parameters	required
result	int	Result code	required
duration_ms	float	Execution duration in milliseconds	required

get_commands abstractmethod

get_commands(session_id: str, command_filter: str | None = None) -> list[CommandRecord]

Retrieve commands for a session.

Parameters:

Name	Type	Description	Default
session_id	str	Session ID	required
command_filter	str | None	Filter by command name (substring match)	None

Returns:

Type	Description
list[CommandRecord]	List of CommandRecord

close abstractmethod

close() -> None

Close database connection and release resources.

FlightSession

Bases: BaseModel

A recorded flight session.

session_id class-attribute instance-attribute

session_id: str = Field(description='Unique session identifier (UUID)')

drone_id class-attribute instance-attribute

drone_id: int = Field(description='Drone ID')

start_time class-attribute instance-attribute

start_time: datetime = Field(description='Session start time')

end_time class-attribute instance-attribute

end_time: datetime | None = Field(default=None, description='Session end time')

notes class-attribute instance-attribute

notes: str | None = Field(default=None, description='Session notes')

Config

from_attributes class-attribute instance-attribute

from_attributes = True

TelemetryRecord

Bases: BaseModel

A single telemetry data point.

id class-attribute instance-attribute

id: int | None = Field(default=None, description='Database record ID')

session_id class-attribute instance-attribute

session_id: str = Field(description='Parent session ID')

timestamp class-attribute instance-attribute

timestamp: datetime = Field(description='Record timestamp')

pos_x class-attribute instance-attribute

pos_x: float = Field(description='X position (cm)')

pos_y class-attribute instance-attribute

pos_y: float = Field(description='Y position (cm)')

pos_z class-attribute instance-attribute

pos_z: float = Field(description='Z position (cm)')

vel_x class-attribute instance-attribute

vel_x: float = Field(description='X velocity (cm/s)')

vel_y class-attribute instance-attribute

vel_y: float = Field(description='Y velocity (cm/s)')

vel_z class-attribute instance-attribute

vel_z: float = Field(description='Z velocity (cm/s)')

yaw class-attribute instance-attribute

yaw: float = Field(description='Yaw angle (degrees)')

pitch class-attribute instance-attribute

pitch: float = Field(description='Pitch angle (degrees)')

roll class-attribute instance-attribute

roll: float = Field(description='Roll angle (degrees)')

altitude class-attribute instance-attribute

altitude: float = Field(description='ToF altitude (cm)')

battery class-attribute instance-attribute

battery: int = Field(description='Battery percentage')

barrier class-attribute instance-attribute

barrier: int = Field(default=0, description='Obstacle detection bitmask')

Config

from_attributes class-attribute instance-attribute

from_attributes = True

from_flight_data classmethod

from_flight_data(session_id: str, data: FlightData) -> TelemetryRecord

Create TelemetryRecord from FlightData model.

CommandRecord

Bases: BaseModel

A recorded command execution.

id class-attribute instance-attribute

id: int | None = Field(default=None, description='Database record ID')

session_id class-attribute instance-attribute

session_id: str = Field(description='Parent session ID')

timestamp class-attribute instance-attribute

timestamp: datetime = Field(description='Command timestamp')

command class-attribute instance-attribute

command: str = Field(description='Command name')

params class-attribute instance-attribute

params: dict[str, Any] = Field(default_factory=dict, description='Command parameters')

result class-attribute instance-attribute

result: int = Field(description='Result code')

duration_ms class-attribute instance-attribute

duration_ms: float = Field(description='Command duration in milliseconds')

Config

from_attributes class-attribute instance-attribute

from_attributes = True

SQLiteLogger

Bases: FlightLogger

SQLite-backed flight logger.

Thread-safe for basic operations. Uses WAL mode for better concurrent read performance.

SCHEMA class-attribute instance-attribute

SCHEMA = '\n    -- Flight sessions table\n    CREATE TABLE IF NOT EXISTS sessions (\n        session_id TEXT PRIMARY KEY,\n        drone_id INTEGER NOT NULL,\n        start_time TEXT NOT NULL,\n        end_time TEXT,\n        notes TEXT\n    );\n\n    -- Telemetry data table\n    CREATE TABLE IF NOT EXISTS telemetry (\n        id INTEGER PRIMARY KEY AUTOINCREMENT,\n        session_id TEXT NOT NULL,\n        timestamp TEXT NOT NULL,\n        pos_x REAL NOT NULL,\n        pos_y REAL NOT NULL,\n        pos_z REAL NOT NULL,\n        vel_x REAL NOT NULL,\n        vel_y REAL NOT NULL,\n        vel_z REAL NOT NULL,\n        yaw REAL NOT NULL,\n        pitch REAL NOT NULL,\n        roll REAL NOT NULL,\n        altitude REAL NOT NULL,\n        battery INTEGER NOT NULL,\n        barrier INTEGER DEFAULT 0,\n        FOREIGN KEY (session_id) REFERENCES sessions(session_id)\n    );\n\n    -- Commands table\n    CREATE TABLE IF NOT EXISTS commands (\n        id INTEGER PRIMARY KEY AUTOINCREMENT,\n        session_id TEXT NOT NULL,\n        timestamp TEXT NOT NULL,\n        command TEXT NOT NULL,\n        params TEXT,\n        result INTEGER NOT NULL,\n        duration_ms REAL NOT NULL,\n        FOREIGN KEY (session_id) REFERENCES sessions(session_id)\n    );\n\n    -- Indexes for common queries\n    CREATE INDEX IF NOT EXISTS idx_telemetry_session ON telemetry(session_id);\n    CREATE INDEX IF NOT EXISTS idx_telemetry_timestamp ON telemetry(timestamp);\n    CREATE INDEX IF NOT EXISTS idx_commands_session ON commands(session_id);\n    CREATE INDEX IF NOT EXISTS idx_sessions_drone ON sessions(drone_id);\n    CREATE INDEX IF NOT EXISTS idx_sessions_start ON sessions(start_time);\n    '

db_path instance-attribute

db_path = Path(db_path)

start_session

start_session(drone_id: int, notes: str | None = None) -> str

Start a new flight session.

end_session

end_session(session_id: str) -> None

End a flight session.

get_session

get_session(session_id: str) -> FlightSession | None

Get session metadata.

list_sessions

list_sessions(drone_id: int | None = None, start_after: datetime | None = None, limit: int = 100) -> list[FlightSession]

List flight sessions.

log_telemetry

log_telemetry(session_id: str, data: FlightData) -> None

Log a telemetry data point.

log_telemetry_batch

log_telemetry_batch(session_id: str, data: list[FlightData]) -> None

Log multiple telemetry data points in batch.

get_telemetry

get_telemetry(session_id: str, start: datetime | None = None, end: datetime | None = None, limit: int = 10000) -> list[TelemetryRecord]

Retrieve telemetry for a session.

log_command

log_command(session_id: str, command: str, params: dict, result: int, duration_ms: float) -> None

Log a command execution.

get_commands

get_commands(session_id: str, command_filter: str | None = None) -> list[CommandRecord]

Retrieve commands for a session.

close

close() -> None

Close database connection.

FileLoggerMiddleware

Thread-safe file logger that writes parsed MAVLink messages to JSONL files.

Files are organized as: logs/drone_YYYY-MM-DD.jsonl Automatically rotates to a new file at midnight.

Usage:

logger = FileLoggerMiddleware("logs")
logger.log_message(msg)  # Called from message analyzer thread

log_dir instance-attribute

log_dir = Path(log_dir)

CENTIDEGREE_FIELDS class-attribute instance-attribute

CENTIDEGREE_FIELDS = {'yaw', 'pitch', 'roll'}

log_message

log_message(msg: Any, state: Any | None = None) -> None

Log a parsed MAVLink message to the current day's log file.

Parameters:

Name	Type	Description	Default
msg	Any	MAVLink message object with get_msg_id() method	required
state	Any | None	Optional State object from StateProcessor	None

close

close() -> None

Close the log file.

CommandLogger

Thread-safe logger for DroneAPI commands.

Logs every API method call with:

• Timestamp
• Method name
• Arguments (positional and keyword)
• Return value or exception
• Execution time

Files are organized as: logs/commands_YYYY-MM-DD.jsonl

Usage:

logger = CommandLogger("logs")

@logger.log
def move(self, direction: Direction, distance: int) -> CommandResult:
    ...

# Or wrap an existing instance:
drone = DroneAPI()
logged_drone = logger.wrap(drone)

log_dir instance-attribute

log_dir = Path(log_dir)

prefix instance-attribute

prefix = prefix

log_args instance-attribute

log_args = log_args

log_result instance-attribute

log_result = log_result

log_exceptions instance-attribute

log_exceptions = log_exceptions

exclude_methods instance-attribute

exclude_methods = exclude_methods or {'get_position', 'get_orientation', 'get_battery', 'get_obstacles', 'get_state', 'get_flight_data'}

enable

enable() -> None

Enable command logging.

disable

disable() -> None

Disable command logging.

log

log(func: Callable[P, R]) -> Callable[P, R]

Decorator to log a function/method call.

Usage:

@logger.log
def move(self, direction: Direction, distance: int):
    ...

log_call

log_call(method_name: str, args: dict[str, Any] | None = None, result: Any = None, success: bool = True, exception: Exception | None = None, elapsed_sec: float | None = None) -> None

Manually log a command call.

Useful when you can't use the decorator.

close

close() -> None

Close the log file.

PostgresLogger

Bases: FlightLogger

PostgreSQL-backed async flight logger.

Uses asyncpg for non-blocking database operations. Ideal for high-frequency telemetry logging.

Usage:

logger = PostgresLogger("postgresql://user:pass@localhost/drone")
await logger.connect()
session_id = await logger.start_session_async(drone_id=1)
await logger.log_telemetry_async(session_id, flight_data)
await logger.close_async()

SCHEMA class-attribute instance-attribute

SCHEMA = '\n    -- Flight sessions table\n    CREATE TABLE IF NOT EXISTS sessions (\n        session_id TEXT PRIMARY KEY,\n        drone_id INTEGER NOT NULL,\n        start_time TIMESTAMPTZ NOT NULL,\n        end_time TIMESTAMPTZ,\n        notes TEXT\n    );\n\n    -- Telemetry data table\n    CREATE TABLE IF NOT EXISTS telemetry (\n        id SERIAL PRIMARY KEY,\n        session_id TEXT NOT NULL REFERENCES sessions(session_id),\n        timestamp TIMESTAMPTZ NOT NULL,\n        pos_x REAL NOT NULL,\n        pos_y REAL NOT NULL,\n        pos_z REAL NOT NULL,\n        vel_x REAL NOT NULL,\n        vel_y REAL NOT NULL,\n        vel_z REAL NOT NULL,\n        yaw REAL NOT NULL,\n        pitch REAL NOT NULL,\n        roll REAL NOT NULL,\n        altitude REAL NOT NULL,\n        battery INTEGER NOT NULL,\n        barrier INTEGER DEFAULT 0\n    );\n\n    -- Commands table\n    CREATE TABLE IF NOT EXISTS commands (\n        id SERIAL PRIMARY KEY,\n        session_id TEXT NOT NULL REFERENCES sessions(session_id),\n        timestamp TIMESTAMPTZ NOT NULL,\n        command TEXT NOT NULL,\n        params JSONB,\n        result INTEGER NOT NULL,\n        duration_ms REAL NOT NULL\n    );\n\n    -- Indexes for common queries\n    CREATE INDEX IF NOT EXISTS idx_telemetry_session ON telemetry(session_id);\n    CREATE INDEX IF NOT EXISTS idx_telemetry_timestamp ON telemetry(timestamp);\n    CREATE INDEX IF NOT EXISTS idx_commands_session ON commands(session_id);\n    CREATE INDEX IF NOT EXISTS idx_sessions_drone ON sessions(drone_id);\n    CREATE INDEX IF NOT EXISTS idx_sessions_start ON sessions(start_time);\n    '

dsn instance-attribute

dsn = dsn

min_connections instance-attribute

min_connections = min_connections

max_connections instance-attribute

max_connections = max_connections

connect async

connect() -> None

Connect to database and initialize schema.

start_session_async async

start_session_async(drone_id: int, notes: str | None = None) -> str

Start a new flight session (async).

end_session_async async

end_session_async(session_id: str) -> None

End a flight session (async).

get_session_async async

get_session_async(session_id: str) -> FlightSession | None

Get session metadata (async).

list_sessions_async async

list_sessions_async(drone_id: int | None = None, start_after: datetime | None = None, limit: int = 100) -> list[FlightSession]

List flight sessions (async).

log_telemetry_async async

log_telemetry_async(session_id: str, data: FlightData) -> None

Log a telemetry data point (async).

log_telemetry_batch_async async

log_telemetry_batch_async(session_id: str, data: list[FlightData]) -> None

Log multiple telemetry data points in batch (async).

get_telemetry_async async

get_telemetry_async(session_id: str, start: datetime | None = None, end: datetime | None = None, limit: int = 10000) -> list[TelemetryRecord]

Retrieve telemetry for a session (async).

log_command_async async

log_command_async(session_id: str, command: str, params: dict, result: int, duration_ms: float) -> None

Log a command execution (async).

get_commands_async async

get_commands_async(session_id: str, command_filter: str | None = None) -> list[CommandRecord]

Retrieve commands for a session (async).

close_async async

close_async() -> None

Close database connection pool (async).

start_session

start_session(drone_id: int, notes: str | None = None) -> str

end_session

end_session(session_id: str) -> None

get_session

get_session(session_id: str) -> FlightSession | None

list_sessions

list_sessions(drone_id: int | None = None, start_after: datetime | None = None, limit: int = 100) -> list[FlightSession]

log_telemetry

log_telemetry(session_id: str, data: FlightData) -> None

log_telemetry_batch

log_telemetry_batch(session_id: str, data: list[FlightData]) -> None

get_telemetry

get_telemetry(session_id: str, start: datetime | None = None, end: datetime | None = None, limit: int = 10000) -> list[TelemetryRecord]

log_command

log_command(session_id: str, command: str, params: dict, result: int, duration_ms: float) -> None

get_commands

get_commands(session_id: str, command_filter: str | None = None) -> list[CommandRecord]

close

close() -> None

create_logging_wrapper

create_logging_wrapper(obj: Any, logger: CommandLogger) -> Any

Create a wrapper that logs all method calls on an object.

This wraps ALL public methods (not starting with _) with logging.

For manual_fly(), automatically injects an on_frame callback to log each individual MANUAL_CONTROL frame sent.

Usage:

drone = DroneAPI()
logged_drone = create_logging_wrapper(drone, CommandLogger("logs"))
logged_drone.takeoff()  # This call is logged
logged_drone.manual_fly(2.0, forward=0.5)  # Logs each frame too