Skip to content

core

Classes:

  • MAVLinkBridge –

    TransportをTopicへ結びつけるためのBridge

  • MAVLinkHistory –

    Topicから受信したMAVLink Messageの履歴を保持するSubscriber

  • MAVLinkPublisher –

    TopicへMAVLink Messageを送信するPublisher

  • MAVLinkRecorder –

    受信したMAVLinkMessageをTlog形式でfileに保存するSubscriber

  • MAVLinkStatus –

    MAVLinkTopicの受信状況を取得するためのclass

  • MAVLinkStatusSnapshot –

    A snapshot of the current status of observed MAVLink messages.

  • MAVLinkSubscriber –

    TopicからMAVLink Messageを受信するSubscriber

  • MAVLinkTopic –

    MAVLink Messageの送受信を媒介するTopic

  • Receiver –

    Transportで受信を行う基底class

  • Sender –

    Transportで送信を行う基底class

  • TLogReader –

    TLog形式で保存されたfileを読み込むReader

  • TopicItem –

    Topic内でやり取りされるデータ

  • TransportBase –

    Transport実装の基底class

MAVLinkBridge 

MAVLinkBridge(transport: TransportBase, topic: MAVLinkTopic, filter: Callable[[TopicItem], bool] | None = None, error_handler: Callable[[Exception], None] | None = None)

TransportをTopicへ結びつけるためのBridge

Parameters:

  • transport

    (TransportBase) –

    BindするTransport

  • topic

    (MAVLinkTopic) –

    BindされるTopic

  • filter

    (Callable[[TopicItem], bool] | None, default: None ) –

    メッセージを受け取るかを判別する関数。Trueを返すと受け取る

  • error_handler

    (Callable[[Exception], None] | None, default: None ) –

    エラー時のcallback関数

Methods:

  • run –

    Bridgeを実行する。stop_eventがsetされると終了する。

Source code in src/mavlink/core.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def __init__(self,transport:TransportBase,topic:MAVLinkTopic,filter: Callable[[TopicItem], bool]|None=None,error_handler: Callable[[Exception], None] | None = None):
    """TransportをTopicへ結びつけるためのBridge

    Args:
        transport (TransportBase): BindするTransport
        topic (MAVLinkTopic): BindされるTopic
        filter (Callable[[TopicItem], bool] | None, optional): メッセージを受け取るかを判別する関数。Trueを返すと受け取る
        error_handler (Callable[[Exception], None] | None, optional): エラー時のcallback関数
    """
    self.transport=transport
    self.reciever=transport.get_receiver()
    self.sender=transport.get_sender()
    self.filter=filter
    self.topic=topic
    self.error_handler=error_handler
    self.mav=mavlink.MAVLink(None)
    self.mav.robust_parsing=True

run 

run(stop_event: Event)

Bridgeを実行する。stop_eventがsetされると終了する。

Parameters:

  • stop_event

    (Event) –

    実行を終了させるためのEvent

Source code in src/mavlink/core.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
def run(self,stop_event:Event):
    """Bridgeを実行する。stop_eventがsetされると終了する。

    Args:
        stop_event (Event): 実行を終了させるためのEvent
    """
    thread_rx=Thread(target=self._run_rx,args=(stop_event,),name="thread_rx")
    thread_tx=Thread(target=self._run_tx,args=(stop_event,),name="thread_tx")
    thread_rx.start()
    thread_tx.start()
    thread_rx.join()
    thread_tx.join()
    self.transport.close()

MAVLinkHistory 

MAVLinkHistory(filter: Callable[[TopicItem], bool], duration: int = 1000000, maxsize: int = None)


              flowchart TD
              mavlink.core.MAVLinkHistory[MAVLinkHistory]
              mavlink.core.MAVLinkSubscriberBase[MAVLinkSubscriberBase]

                              mavlink.core.MAVLinkSubscriberBase --> mavlink.core.MAVLinkHistory
                


              click mavlink.core.MAVLinkHistory href "" "mavlink.core.MAVLinkHistory"
              click mavlink.core.MAVLinkSubscriberBase href "" "mavlink.core.MAVLinkSubscriberBase"

Topicから受信したMAVLink Messageの履歴を保持するSubscriber

Returned by:

Methods:

  • clear –

    Historyをclearする

  • items –

    保持しているHistoryを取得する

  • latest –

    保持している最新のメッセージを取得する

  • sync –

    MAVLink MessageのHistoryと最新のメッセージを更新する

Source code in src/mavlink/core.py 
72
73
74
75
76
def __init__(self, filter:Callable[[TopicItem],bool],duration:int=1000_000,maxsize:int=None):
    super().__init__(filter=filter)
    self.__duration = duration
    self.__queue : Queue[TopicItem] = Queue(maxsize=maxsize)
    self.history : deque[TopicItem] = deque(maxlen=maxsize)

clear 

clear()

Historyをclearする

Source code in src/mavlink/core.py 
101
102
103
104
def clear(self):
    """Historyをclearする"""
    self.history.clear()
    self.__queue.queue.clear()

items 

items() -> list[TopicItem]

保持しているHistoryを取得する

Returns:

Source code in src/mavlink/core.py 
 94
 95
 96
 97
 98
 99
100
def items(self) -> list[TopicItem]:
    """保持しているHistoryを取得する

    Returns:
        list[TopicItem]: 保持しているHistory
    """
    return self.history

latest 

latest() -> Optional[TopicItem]

保持している最新のメッセージを取得する

Returns:

  • Optional[TopicItem] –

    Optional[TopicItem]: 保持している最新のメッセージ。何もメッセージを受け取っていなければNoneを返す。

Source code in src/mavlink/core.py 
105
106
107
108
109
110
111
112
113
def latest(self) -> Optional[TopicItem]:
    """保持している最新のメッセージを取得する

    Returns:
        Optional[TopicItem]: 保持している最新のメッセージ。何もメッセージを受け取っていなければNoneを返す。
    """
    if self.history:
        return self.history[-1]
    return None

sync 

sync(sync_timestamp: Optional[int] = None)

MAVLink MessageのHistoryと最新のメッセージを更新する

Parameters:

  • sync_timestamp

    (Optional[int], default: None ) –

    Historyが保持するメッセージの時刻の基準値。Noneを指定すると最新のメッセージのtimestampを基準とする。 Defaults to None.

Source code in src/mavlink/core.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def sync(self,sync_timestamp:Optional[int]=None):
    """MAVLink MessageのHistoryと最新のメッセージを更新する

    Args:
        sync_timestamp (Optional[int], optional): Historyが保持するメッセージの時刻の基準値。Noneを指定すると最新のメッセージのtimestampを基準とする。 Defaults to None.
    """
    while not self.__queue.empty():
        item = self.__queue.get()
        self.history.append(item)
    if len(self.history) == 0:
        return
    if sync_timestamp is None:
        sync_timestamp = self.history[-1].timestamp if self.history else 0.0
    while self.history and (sync_timestamp - self.history[0].timestamp) > self.__duration:
        self.history.popleft()

MAVLinkPublisher 

MAVLinkPublisher(topic: MAVLinkTopic, source_id: str)

TopicへMAVLink Messageを送信するPublisher

Returned by:

Methods:

  • publish –

    Topicへメッセージをpublishする

Source code in src/mavlink/core.py 
25
26
27
def __init__(self,topic:MAVLinkTopic,source_id:str):
    self.topic=topic
    self.source_id=source_id

publish 

publish(timestamp: int, message: MAVLink_message) -> None

Topicへメッセージをpublishする

Parameters:

  • timestamp

    (int) –

    メッセージのtimestamp

  • message

    (MAVLink_message) –

    送信するメッセージ

Source code in src/mavlink/core.py 
28
29
30
31
32
33
34
35
def publish(self,timestamp: int,message: mavlink.MAVLink_message) -> None:
    """Topicへメッセージをpublishする

    Args:
        timestamp (int): メッセージのtimestamp
        message (mavlink.MAVLink_message): 送信するメッセージ
    """
    self.topic._publish(timestamp, message,self.source_id)

MAVLinkRecorder 

MAVLinkRecorder(filepath: Path)


              flowchart TD
              mavlink.core.MAVLinkRecorder[MAVLinkRecorder]
              mavlink.core.MAVLinkSubscriberBase[MAVLinkSubscriberBase]

                              mavlink.core.MAVLinkSubscriberBase --> mavlink.core.MAVLinkRecorder
                


              click mavlink.core.MAVLinkRecorder href "" "mavlink.core.MAVLinkRecorder"
              click mavlink.core.MAVLinkSubscriberBase href "" "mavlink.core.MAVLinkSubscriberBase"

受信したMAVLinkMessageをTlog形式でfileに保存するSubscriber

Returned by:
Source code in src/mavlink/core.py 
117
118
119
120
121
122
123
def __init__(self, filepath:Path):
    super().__init__(lambda _item:True)
    self.file = filepath
    self.file.parent.mkdir(parents=True, exist_ok=True)
    self.file.touch()
    self.lock=Lock()
    self.parser=struct.Struct(">Q")

MAVLinkStatus 

MAVLinkStatus()

MAVLinkTopicの受信状況を取得するためのclass

Returned by:

Methods:

  • snapshot –

    現在の受信状況を取得する

Source code in src/mavlink/core.py 
184
185
186
187
def __init__(self):
    self.observed_messages: set[tuple[int, int, int]] = set()
    self.last_received: dict[tuple[int, int, int], int] = {}
    self._lock = Lock()  # Lock for thread-safe operations

snapshot 

snapshot() -> MAVLinkStatusSnapshot

現在の受信状況を取得する

Returns:

Source code in src/mavlink/core.py 
192
193
194
195
196
197
198
199
200
201
202
def snapshot(self) -> MAVLinkStatusSnapshot:
    """現在の受信状況を取得する

    Returns:
        MAVLinkStatusSnapshot: 現在の受信状況のsnapshot
    """
    with self._lock:
        return MAVLinkStatusSnapshot(
            observed_messages=frozenset(self.observed_messages),
            last_received=self.last_received.copy()
        )

MAVLinkStatusSnapshot dataclass 

MAVLinkStatusSnapshot(observed_messages: frozenset[tuple[int, int, int]], last_received: dict[tuple[int, int, int], int])

A snapshot of the current status of observed MAVLink messages.

Returned by:

MAVLinkSubscriber 

MAVLinkSubscriber(filter: Callable[[TopicItem], bool], maxsize: int = 10000)


              flowchart TD
              mavlink.core.MAVLinkSubscriber[MAVLinkSubscriber]
              mavlink.core.MAVLinkSubscriberBase[MAVLinkSubscriberBase]

                              mavlink.core.MAVLinkSubscriberBase --> mavlink.core.MAVLinkSubscriber
                


              click mavlink.core.MAVLinkSubscriber href "" "mavlink.core.MAVLinkSubscriber"
              click mavlink.core.MAVLinkSubscriberBase href "" "mavlink.core.MAVLinkSubscriberBase"

TopicからMAVLink Messageを受信するSubscriber

Returned by:

Methods:

  • get –

    timeoutで指定した時間内でTopicからメッセージを受け取る

Source code in src/mavlink/core.py 
50
51
52
def __init__(self, filter:Callable[[TopicItem],bool],maxsize:int=10000):
    super().__init__(filter=filter)
    self.__queue : Queue[TopicItem] = Queue(maxsize=maxsize)

get 

get(timeout: Optional[float] = None) -> Optional[TopicItem]

timeoutで指定した時間内でTopicからメッセージを受け取る

Parameters:

  • timeout

    (Optional[float], default: None ) –

    Timeoutするまでの時間。 Defaults to None.

Returns:

  • Optional[TopicItem] –

    Optional[TopicItem]: 受信したメッセージ。指定された時間内に受信できなければNoneが返される。

Source code in src/mavlink/core.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def get(self,timeout:Optional[float]=None) -> Optional[TopicItem]:
    """timeoutで指定した時間内でTopicからメッセージを受け取る

    Args:
        timeout (Optional[float], optional): Timeoutするまでの時間。 Defaults to None.

    Returns:
        Optional[TopicItem]: 受信したメッセージ。指定された時間内に受信できなければNoneが返される。
    """
    try:
        item = self.__queue.get(timeout=timeout)
        return item
    except Empty:
        return None

MAVLinkTopic 

MAVLinkTopic()

MAVLink Messageの送受信を媒介するTopic

Used by:

Methods:

Source code in src/mavlink/core.py 
331
332
333
334
335
336
337
def __init__(self):
    """MAVLink Messageの送受信を媒介するTopic
    """
    self.subscribers : set[MAVLinkSubscriberBase] = set()
    self.lock : Lock = Lock()
    self.logger = getLogger(__name__)
    self.status = MAVLinkStatus()  # Initialize a single MAVLinkStatus instance for tracking message status

create_history_subscriber 

create_history_subscriber(filter: Callable[[TopicItem], bool], duration: int = 1000000, maxsize: int = 1000) -> MAVLinkHistory

MAVLinkHistorySubscriberのインスタンスを作成する

Parameters:

  • filter

    (Callable[[TopicItem], bool]) –

    メッセージを受け取るかを判別する関数。Trueを返すと受け取る

  • duration

    (int, default: 1000000 ) –

    Historyで保持するMessageの時間幅。単位はus

  • maxsize

    (int, default: 1000 ) –

    Historyの内部的に用いるdequeの長さ

Returns:

Source code in src/mavlink/core.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def create_history_subscriber(self,filter:Callable[[TopicItem],bool],duration:int=1000_000,maxsize:int=1000) -> MAVLinkHistory:
    """MAVLinkHistorySubscriberのインスタンスを作成する

    Args:
        filter (Callable[[TopicItem],bool]): メッセージを受け取るかを判別する関数。Trueを返すと受け取る
        duration (int, optional): Historyで保持するMessageの時間幅。単位はus
        maxsize (int, optional): Historyの内部的に用いるdequeの長さ

    Returns:
        MAVLinkHistory
    """
    history_subscriber = MAVLinkHistory(filter,duration=duration,maxsize=maxsize)
    with self.lock:
        self.subscribers.add(history_subscriber)
    return history_subscriber

create_publisher 

create_publisher(source_id: str) -> MAVLinkPublisher

MAVLinkPublisherのインスタンスを作成する

Parameters:

  • source_id

    (str) –

    送信元を判別するためのID

Returns:

Source code in src/mavlink/core.py 
393
394
395
396
397
398
399
400
401
402
def create_publisher(self, source_id:str) -> MAVLinkPublisher:
    """MAVLinkPublisherのインスタンスを作成する

    Args:
        source_id (str): 送信元を判別するためのID

    Returns:
        MAVLinkPublisher
    """
    return MAVLinkPublisher(self, source_id=source_id)

create_record 

create_record(filepath: Path) -> MAVLinkRecorder

MAVLinkRecorderのインスタンスを作成する

Parameters:

  • filepath

    (Path) –

    保存先のfileへのpath

Returns:

Source code in src/mavlink/core.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
def create_record(self,filepath:Path)->MAVLinkRecorder:
    """MAVLinkRecorderのインスタンスを作成する

    Args:
        filepath (Path): 保存先のfileへのpath

    Returns:
        MAVLinkRecorder
    """
    recorder = MAVLinkRecorder(filepath=filepath)
    with self.lock:
        self.subscribers.add(recorder)
    return recorder

create_subscriber 

create_subscriber(filter: Callable[[TopicItem], bool], maxsize: int = 100) -> MAVLinkSubscriber

Subscriberのインスタンスを作成する

Parameters:

  • filter

    (Callable[[TopicItem], bool]) –

    メッセージを受け取るかを判別する関数。Trueを返すと受け取る

  • maxsize

    (int, default: 100 ) –

    内部的に用いるqueueのサイズ

Returns:

Source code in src/mavlink/core.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def create_subscriber(self, filter:Callable[[TopicItem],bool],maxsize:int=100) -> MAVLinkSubscriber:
    """Subscriberのインスタンスを作成する

    Args:
        filter (Callable[[TopicItem],bool]): メッセージを受け取るかを判別する関数。Trueを返すと受け取る
        maxsize (int, optional): 内部的に用いるqueueのサイズ

    Returns:
        MAVLinkSubscriber
    """
    subscriber = MAVLinkSubscriber(filter,maxsize=maxsize)
    with self.lock:
        self.subscribers.add(subscriber)
    return subscriber

get_status 

get_status() -> MAVLinkStatus

MAVLinkStatusのインスタンスを作成する

Returns:

Source code in src/mavlink/core.py 
403
404
405
406
407
408
409
def get_status(self) -> MAVLinkStatus:
    """MAVLinkStatusのインスタンスを作成する

    Returns:
        MAVLinkStatus
    """
    return self.status

unsubscribe 

unsubscribe(subscriber: MAVLinkSubscriberBase)

Subscriberの登録を解除する

Parameters:

  • subscriber

    (MAVLinkSubscriberBase) –

    登録を解除するSubscriber

Source code in src/mavlink/core.py 
380
381
382
383
384
385
386
387
def unsubscribe(self, subscriber:MAVLinkSubscriberBase):
    """Subscriberの登録を解除する

    Args:
        subscriber (MAVLinkSubscriberBase): 登録を解除するSubscriber
    """
    with self.lock:
        self.subscribers.discard(subscriber)

Receiver 


              flowchart TD
              mavlink.core.Receiver[Receiver]

              

              click mavlink.core.Receiver href "" "mavlink.core.Receiver"

Transportで受信を行う基底class

Returned by:

Methods:

  • recv –

    バイト列を受信する

recv abstractmethod 

recv(timeout: float) -> Sequence[int] | None

バイト列を受信する

Parameters:

  • timeout

    (float) –

    受信でtimeoutするまでの時間。単位はs

Returns:

  • Sequence[int] | None –

    Sequence[int] | None: 受信したバイト列。何も受信しなかった場合はNoneを返す

Source code in src/mavlink/core.py 
217
218
219
220
221
222
223
224
225
226
227
@abc.abstractmethod
def recv(self,timeout:float)->Sequence[int] | None:
    """バイト列を受信する

    Args:
        timeout (float): 受信でtimeoutするまでの時間。単位はs

    Returns:
        Sequence[int] | None: 受信したバイト列。何も受信しなかった場合はNoneを返す
    """
    ...

Sender 


              flowchart TD
              mavlink.core.Sender[Sender]

              

              click mavlink.core.Sender href "" "mavlink.core.Sender"

Transportで送信を行う基底class

Returned by:

Methods:

  • send –

    バイト列の送信を行う

send abstractmethod 

send(data: Sequence[int])

バイト列の送信を行う

Parameters:

Source code in src/mavlink/core.py 
206
207
208
209
210
211
212
213
@abc.abstractmethod
def send(self,data:Sequence[int]):
    """バイト列の送信を行う

    Args:
        data (Sequence[int]): 送信するバイト列
    """
    ...

TLogReader 

TLogReader(filepath: Path)

TLog形式で保存されたfileを読み込むReader

Example 
reader=TLogReader(Path("sample.tlog")) # sample.tlogを読み込む
for (timestamp,message) in reader:
    print(f"{timestamp=},{message=}") # Iteratorでtimestampとmessageを取り出す

Parameters:

  • filepath

    (Path) –

    読み込むfileへのpath

Source code in src/mavlink/core.py 
143
144
145
146
147
148
149
150
151
152
153
def __init__(self,filepath:Path):
    """TLog形式で保存されたfileを読み込むReader

    Args:
        filepath (Path): 読み込むfileへのpath
    """
    with filepath.open("rb") as f:
        self.buffer=f.read()
    self.idx=0
    self.mav=mavlink.MAVLink(None)
    self.parser=struct.Struct(">Q")

TopicItem dataclass 

TopicItem(timestamp: int, message: MAVLink_message, source_id: str)

Topic内でやり取りされるデータ

Returned by: Used by:

TransportBase 


              flowchart TD
              mavlink.core.TransportBase[TransportBase]

              

              click mavlink.core.TransportBase href "" "mavlink.core.TransportBase"

Transport実装の基底class

Used by:

Methods:

close abstractmethod 

close()

通信の終了処理を行う

Source code in src/mavlink/core.py 
254
255
256
257
258
@abc.abstractmethod
def close(self):
    """通信の終了処理を行う
    """
    ...

get_receiver abstractmethod 

get_receiver() -> Receiver | None

Receiverを返す

Returns:

  • Receiver | None –

    Receiver | None: Receiverが存在しなければNoneを返す

Source code in src/mavlink/core.py 
246
247
248
249
250
251
252
253
@abc.abstractmethod
def get_receiver(self)->Receiver | None:
    """Receiverを返す

    Returns:
        Receiver | None: Receiverが存在しなければNoneを返す
    """
    ...

get_sender abstractmethod 

get_sender() -> Sender | None

Senderを返す

Returns:

  • Sender | None –

    Sender | None: Senderが存在しなければNoneを返す

Source code in src/mavlink/core.py 
238
239
240
241
242
243
244
245
@abc.abstractmethod
def get_sender(self)->Sender | None:
    """Senderを返す

    Returns:
        Sender | None: Senderが存在しなければNoneを返す
    """
    ...

get_source_id abstractmethod 

get_source_id() -> str

Publisherの送信元を判別するためのIDを返す

Returns:

  • str ( str ) –

    送信元を判別するためのID

Source code in src/mavlink/core.py 
230
231
232
233
234
235
236
237
@abc.abstractmethod
def get_source_id(self)->str:
    """Publisherの送信元を判別するためのIDを返す

    Returns:
        str: 送信元を判別するためのID
    """
    ...