State Vector
This module implements a threaded vehicle state management system with: - Continuous state vector updates - Position/orientation simulation - Battery consumption modeling - Thread-safe queue communication
Classes
- StateVector
Threaded vehicle state manager
- Position
Geographic position tracker
- Orientation
Vehicle attitude simulator
- Battery
Battery consumption model
Functions
- start_state_vector
Initialize and start state vector thread
- class state_vector.StateVector(vehicle_id: int, latitude, longitude, action_queue, topic_queue)[source]
Bases:
ThreadThreaded vehicle state manager with queue-based communication.
- Parameters:
vehicle_id (int) – Unique vehicle identifier
latitude (float) – Initial latitude in decimal degrees
longitude (float) – Initial longitude in decimal degrees
action_queue (queue.Queue) – Input queue for command updates (consumer)
topic_queue (queue.Queue) – Output queue for state publications (producer)
- Variables:
vehicle_status (str) – Current operational status (‘Available’|’Busy’|’Charging’)
battery (Battery) – Battery state model
position (Position) – Geographic position tracker
orientation (Orientation) – Vehicle attitude simulation
- __init__(vehicle_id: int, latitude, longitude, action_queue, topic_queue)[source]
This constructor should always be called with keyword arguments. Arguments are:
group should be None; reserved for future extension when a ThreadGroup class is implemented.
target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.
name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.
args is a list or tuple of arguments for the target invocation. Defaults to ().
kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.
context is the contextvars.Context value to use for the thread. The default value is None, which means to check sys.flags.thread_inherit_context. If that flag is true, use a copy of the context of the caller. If false, use an empty context. To explicitly start with an empty context, pass a new instance of contextvars.Context(). To explicitly start with a copy of the current context, pass the value from contextvars.copy_context().
If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.
- run()[source]
Main state update loop with queue processing.
Publishes initial state
Processes incoming commands from action_queue
Updates simulation values
Publishes updated state to topic_queue
Note
Runs continuously until thread termination
- class state_vector.Position(latitude, longitude)[source]
Bases:
objectGeographic position tracker with altitude simulation.
- Parameters:
latitude (float) – Initial latitude in decimal degrees
longitude (float) – Initial longitude in decimal degrees
- class state_vector.Orientation[source]
Bases:
objectVehicle attitude simulator with random walk model.
- class state_vector.Battery(battery_capacity: float, battery_percentage: float)[source]
Bases:
objectBattery consumption model with linear discharge.
- Parameters:
battery_capacity (float) – Total capacity in watt-hours
battery_percentage (float) – Initial charge percentage
- state_vector.start_state_vector(vehicle_id: int, longitude, latitude, action_queue, topic_queue)[source]
Initialize and start state vector management thread.
- Parameters:
vehicle_id (int) – Unique vehicle identifier
longitude (float) – Initial longitude in decimal degrees
latitude (float) – Initial latitude in decimal degrees
action_queue (queue.Queue) – Command input queue
topic_queue (queue.Queue) – State output queue
Example
Start state tracking for drone 5:
action_q = queue.Queue() topic_q = queue.Queue() start_state_vector(5, 40.7128, -74.0060, action_q, topic_q)