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: Thread

Threaded 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.

  1. Publishes initial state

  2. Processes incoming commands from action_queue

  3. Updates simulation values

  4. Publishes updated state to topic_queue

Note

Runs continuously until thread termination

update(vehicle_status: str, longitude, latitude)[source]

Update vehicle state with new command parameters.

Parameters:
  • vehicle_status (str) – New operational status

  • longitude (float) – Target longitude in decimal degrees

  • latitude (float) – Target latitude in decimal degrees

get_values()[source]

Package current state into dictionary format.

Returns:

Complete vehicle state snapshot

Return type:

dict

class state_vector.Position(latitude, longitude)[source]

Bases: object

Geographic position tracker with altitude simulation.

Parameters:
  • latitude (float) – Initial latitude in decimal degrees

  • longitude (float) – Initial longitude in decimal degrees

__init__(latitude, longitude)[source]
update_position(latitude, longitude)[source]
class state_vector.Orientation[source]

Bases: object

Vehicle attitude simulator with random walk model.

__init__()[source]
update_orientation()[source]
class state_vector.Battery(battery_capacity: float, battery_percentage: float)[source]

Bases: object

Battery consumption model with linear discharge.

Parameters:
  • battery_capacity (float) – Total capacity in watt-hours

  • battery_percentage (float) – Initial charge percentage

__init__(battery_capacity: float, battery_percentage: float)[source]
update_battery()[source]
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)