beatzxbt

Aviation system state management

Exported on 12/22/2025 at 17:46:13 GMT from Cursor (2.2.43)

---

User

Im having a bit of a mess with handling and managing all the various components' states in this system.

We have Airport/Aircraft states as the absolute bottom layer, then Flight/Route states on top of these, then Fleets/Schedules as the absolute top (being purely a state-changing machine).

beatzxbt

cimport numpy as cnp
from libc.stdint cimport uint64_t as u64

ctypedef fused uint_t:
    cnp.uint8_t
    cnp.uint16_t
    cnp.uint32_t
    cnp.uint64_t

ctypedef fused int_t:

beatzxbt

import asyncio
from libc.stdint cimport (
    UINT8_MAX,
    uint8_t as u8,
)

cdef class BytesRingBuffer:
    """A fixed-size ring buffer for bytes objects"""
    def __cinit__(self, bint only_insert_unique=False, u8 max_size=UINT8_MAX):
        """

beatzxbt

import numpy as np

cimport numpy as cnp
from libc.stdint cimport uint64_t, int64_t


cdef class RingBufferOneDim:
    """
    A 1-dimensional fixed-size circular buffer for floats/doubles.
    """

beatzxbt

from libc.stdint cimport uint32_t

from mm_toolbox.time.time cimport time_s

from .engine cimport OrderAction

cdef class RateLimitCounter:
    cdef:
        uint32_t    tokens_per_sec
        uint32_t    tokens_remaining

beatzxbt

import numpy as np
cimport numpy as np
from libc.stdint cimport uint16_t
from libc.stdlib cimport strtod
from cpython.unicode cimport PyUnicode_AsUTF8

cpdef np.ndarray[np.double_t, ndim=2] parse_list_list_str_to_floats(list[list[str]] data):
    """
    Parses a list of lists of strings into a NumPy array of floats.

beatzxbt

import xxhash
import asyncio
import platform
from collections import deque
from warnings import warn as warning
from typing import List, Dict, Optional

from mm_toolbox.time import time_ns, time_s

from .conn.raw cimport WsConnection

beatzxbt

import numpy as np
cimport numpy as np
from libc.stdlib cimport strtod
from cpython.unicode cimport PyUnicode_AsUTF8
from cython cimport boundscheck, wraparound

@boundscheck(False)
@wraparound(False)
cdef np.ndarray[np.float64_t, ndim=2] parse_list_strings_to_floats(list[list[str]] data):
    """

beatzxbt

import orjson
import asyncio
from warnings import warn as warning
from dataclasses import dataclass
from aiohttp import ClientSession, ClientWebSocketResponse, WSMsgType
from typing import List, Dict, Callable, Any, Optional

from mm_toolbox.src.logging import Logger
from mm_toolbox.src.time import time_ms
from mm_toolbox.src.ringbuffer import RingBufferSingleDimFloat

beatzxbt

def generate_new_buffer(size: float, strength: float) -> tuple[float, float]:
    """
    Generates a new buffer range based on the size and strength.
    """
    return (size * (1 - strength), size * (1 + strength))

def process_data(data) -> None:
    """
    Processes the data with varying buffer strengths and prints the results.
    """