Skip to content

blueye.sdk.logs

logs

Classes:

  • LegacyLogFile –

    This class is a container for a log file stored on the drone

  • LegacyLogs –

    This class is an index of the legacy csv log files stored on the drone

  • LogFile –
  • LogStream –

    Class for streaming a log

  • Logs –

Functions:

LegacyLogFile 

LegacyLogFile(maxdepth, name, timestamp, binsize, ip)

This class is a container for a log file stored on the drone

The drone lists the file name, max depth, start time, and file size for each log, and you can show this information by printing the log object, eg. on a Drone object called myDrone:

print(myDrone.logs[0])

or, if you want to display the header you can format the object with with_header:

print(f"{myDrone.logs[0]:with_header}")

Calling the download() method on a log object will pull the CSV (Comma Separated Value) file from the drone to your local filesystem.

Methods:

  • download –

    Download the specified log to your local file system

Source code in blueye/sdk/logs.py 
304
305
306
307
308
309
310
311
312
313
314
315
def __init__(self, maxdepth, name, timestamp, binsize, ip):
    self.maxdepth = maxdepth
    self.name = name
    self.timestamp: datetime = dateutil.parser.isoparse(timestamp)
    self.binsize = binsize
    self.download_path = "http://" + ip + "/logcsv/" + name
    self._formatted_values = [
        self.name,
        self.timestamp.strftime("%d. %b %Y %H:%M"),
        f"{self.maxdepth/1000:.2f} m",
        human_readable_filesize(self.binsize),
    ]

download 

download(
    output_path=None,
    output_name=None,
    downsample_divisor=10,
)

Download the specified log to your local file system

If you specify an output_path the log file will be downloaded to that directory instead of the current one.

Specifying output_name will overwrite the default file name with whatever you have specified (be sure to include the .csv extension).

The drone samples the log content at 10 Hz, and by default this function downsamples this rate to 1 Hz.

Source code in blueye/sdk/logs.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def download(self, output_path=None, output_name=None, downsample_divisor=10):
    """
    Download the specified log to your local file system

    If you specify an output_path the log file will be downloaded to that directory
    instead of the current one.

    Specifying output_name will overwrite the default file name with whatever you
    have specified (be sure to include the .csv extension).

    The drone samples the log content at 10 Hz, and by default this function downsamples this
    rate to 1 Hz.
    """
    log = requests.get(self.download_path, params={"divisor": downsample_divisor}).content
    if output_path is None:
        output_path = "./"
    if output_name is None:
        output_name = self.name
    with open(f"{output_path}{output_name}", "wb") as f:
        f.write(log)

LegacyLogs 

LegacyLogs(parent_drone, auto_download_index=False)

This class is an index of the legacy csv log files stored on the drone

To show the available logs you simply print this object, ie. if your Drone object is called myDrone, you can do:

print(myDrone.legacy_logs)

This will print a list of all available logs, with some of their metadata, such as name and maxdepth.

You can access logfile objects either by index or by name. Eg. if you want the first logfile in the list you can do myDrone.logs[0], or if you want some particular log you can do myDrone.logs["exampleName0001.csv"]. You can even give it a slice, so if you want the last 10 logs you can do myDrone.logs[-10:].

Methods:

Source code in blueye/sdk/logs.py 
372
373
374
375
376
377
378
379
def __init__(self, parent_drone, auto_download_index=False):
    self.ip = parent_drone._ip
    self._parent_drone = parent_drone
    self.index_downloaded = False
    if auto_download_index:
        self.refresh_log_index()
    else:
        self._logs = {}

refresh_log_index 

refresh_log_index(get_all_logs=False)

Refresh the log index from the drone

This is method is run on the first log access by default, but if you would like to check for new log files it can be called at any time.

Pass with get_all_logs=True to include logs that are not classified as dives.

Source code in blueye/sdk/logs.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def refresh_log_index(self, get_all_logs=False):
    """Refresh the log index from the drone

    This is method is run on the first log access by default, but if you would like to check
    for new log files it can be called at any time.

    Pass with `get_all_logs=True` to include logs that are not classified as dives.
    """
    if not self._parent_drone.connected:
        raise ConnectionError(
            "The connection to the drone is not established, try calling the connect method "
            "before retrying"
        )
    list_of_logs_in_dictionaries = self._get_list_of_logs_from_drone(get_all_logs)
    self._logs = self._build_log_files_from_dictionary(list_of_logs_in_dictionaries)
    self.index_downloaded = True

LogFile 

LogFile(
    name: str,
    is_dive: bool,
    filesize: int,
    start_time: int,
    max_depth_magnitude: int,
    ip: str,
)

Methods:

Source code in blueye/sdk/logs.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    name: str,
    is_dive: bool,
    filesize: int,
    start_time: int,
    max_depth_magnitude: int,
    ip: str,
):
    self.name = name
    self.is_dive = is_dive
    self.filesize = filesize
    self.start_time: datetime = datetime.fromtimestamp(start_time, tz=timezone.utc)
    self.max_depth_magnitude = max_depth_magnitude
    self.download_url = f"http://{ip}/logs/{self.name}/binlog"
    self.content = None
    self._formatted_values = [
        self.name,
        self.start_time.strftime("%d. %b %Y %H:%M"),
        f"{self.max_depth_magnitude} m",
        human_readable_filesize(self.filesize),
    ]

download 

download(
    output_path: Optional[Path | str] = None,
    write_to_file: bool = True,
    timeout: float = 1,
    overwrite_cache: bool = False,
) -> bytes

Download a log file from the drone

Arguments:

  • output_path: Path to write the log file to. If None, the log will be written to the current working directory. If the path is a directory, the log will be downloaded to that directory with its original name. Else the log will be downloaded to the specified path.
  • write_to_file: If True, the log will be written to the specified path. If False, the log will only be returned as a bytes object.
  • timeout: Seconds to wait for response
  • overwrite_cache: If True, the log will be downloaded even if it is already been downloaded.

Returns:

The compressed log file as a bytes object.

Source code in blueye/sdk/logs.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def download(
    self,
    output_path: Optional[Path | str] = None,
    write_to_file: bool = True,
    timeout: float = 1,
    overwrite_cache: bool = False,
) -> bytes:
    """Download a log file from the drone

    *Arguments*:

    * `output_path`:
        Path to write the log file to. If `None`, the log will be written to the
        current working directory. If the path is a directory, the log will be
        downloaded to that directory with its original name. Else the log will be
        downloaded to the specified path.
    * `write_to_file`:
        If True, the log will be written to the specified path. If False, the
        log will only be returned as a bytes object.
    * `timeout`:
        Seconds to wait for response
    * `overwrite_cache`:
        If True, the log will be downloaded even if it is already been downloaded.

    *Returns*:

    The compressed log file as a bytes object.
    """
    if self.content is None or overwrite_cache:
        self.content = requests.get(self.download_url, timeout=timeout).content
    if write_to_file:
        if output_path is None:
            output_path = Path(f"{self.name}.bez")
        else:
            if type(output_path) == str:
                output_path = Path(output_path)
            if output_path.is_dir():
                output_path = output_path.joinpath(f"{self.name}.bez")
        with open(output_path, "wb") as f:
            f.write(self.content)
    return self.content

parse_to_stream 

parse_to_stream() -> LogStream

Parse the log file to a stream

Will download the log if it is not already downloaded.

Returns:

A LogStream object

Source code in blueye/sdk/logs.py 
151
152
153
154
155
156
157
158
159
160
def parse_to_stream(self) -> LogStream:
    """Parse the log file to a stream

    Will download the log if it is not already downloaded.

    *Returns*:

    A `LogStream` object
    """
    return LogStream(self.download(write_to_file=False))

LogStream 

LogStream(log: bytes, decompress: bool = True)

Class for streaming a log

Creates a stream from a downloaded log file. Iterate over the object to get the next log record.

Source code in blueye/sdk/logs.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, log: bytes, decompress: bool = True) -> Iterator[
    Tuple[
        proto.datetime_helpers.DatetimeWithNanoseconds,  # Real time clock
        timedelta,  # Time since first message
        proto.message.MessageMeta,  # Message type
        proto.message.Message,  # Message contents
    ]
]:
    if decompress:
        self.decompressed_log = decompress_log(log)
    else:
        self.decompressed_log = log
    self.pos = 0
    self.start_monotonic: proto.datetime_helpers.DatetimeWithNanoseconds = 0

Logs 

Logs(parent_drone, auto_download_index=False)

Methods:

  • filter –

    Return a new Logs object with only those matching the filter

  • refresh_log_index –

    Refresh the log index from the drone

Source code in blueye/sdk/logs.py 
178
179
180
181
182
183
184
def __init__(self, parent_drone, auto_download_index=False):
    self._parent_drone = parent_drone
    self.auto_download_index = auto_download_index
    self.index_downloaded = False
    self._logs = {}
    if auto_download_index:
        self.refresh_log_index()

filter 

filter(filter_func: Callable[[LogFile], bool]) -> Logs

Return a new Logs object with only those matching the filter

Eg. to get logs classified as a dive: 

dive_logs = myDrone.logs.filter(lambda log: log.is_dive)

or to get all logs with a max depth greater than 10m: 

deep_logs = myDrone.logs.filter(lambda log: log.max_depth_magnitude > 10)

Source code in blueye/sdk/logs.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def filter(self, filter_func: Callable[[LogFile], bool]) -> Logs:
    """Return a new Logs object with only those matching the filter

    Eg. to get logs classified as a dive:
    ```
    dive_logs = myDrone.logs.filter(lambda log: log.is_dive)
    ```

    or to get all logs with a max depth greater than 10m:
    ```
    deep_logs = myDrone.logs.filter(lambda log: log.max_depth_magnitude > 10)
    ```
    """
    if not self.index_downloaded:
        self.refresh_log_index()
    filtered_logs = Logs(self._parent_drone)
    filtered_logs.index_downloaded = True
    for log in self:
        if filter_func(log):
            filtered_logs._logs[log.name] = log
    return filtered_logs

refresh_log_index 

refresh_log_index()

Refresh the log index from the drone

This is method is run on the first log access by default, but if you would like to check for new log files it can be called at any time.

Source code in blueye/sdk/logs.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def refresh_log_index(self):
    """Refresh the log index from the drone

    This is method is run on the first log access by default, but if you would like to check
    for new log files it can be called at any time.
    """
    if not self._parent_drone.connected:
        raise ConnectionError(
            "The connection to the drone is not established, try calling the connect method "
            "before retrying"
        )
    logger.debug("Refreshing log index")
    logs_endpoint = f"http://{self._parent_drone._ip}/logs"
    logs: List[dict] = requests.get(logs_endpoint).json()

    if version.parse(self._parent_drone.software_version_short) < version.parse("3.3"):
        # Extend index with dive info, sends a request for each log file so can be quite slow
        # for drones with many logs. Not necessary for Blunux >= 3.3 as dive info is included in
        # the index.
        logger.debug(f"Getting dive info for {len(logs)} logs")
        for index, log in enumerate(logs):
            dive_info = requests.get(f"{logs_endpoint}/{log['name']}/dive_info").json()
            logs[index].update(dive_info)

    # Instantiate log objects for each log
    logger.debug(f"Creating log objects for {len(logs)} logs")
    for log in logs:
        if log["has_binlog"]:
            self._logs[log["name"]] = LogFile(
                log["name"],
                log["is_dive"],
                log["binlog_size"],
                log["start_time"],
                log["max_depth_magnitude"],
                self._parent_drone._ip,
            )
        else:
            logger.info(f"Log {log['name']} does not have a binlog, ignoring")
    self.index_downloaded = True

decompress_log 

decompress_log(log: bytes) -> bytes

Decompress a log file

Source code in blueye/sdk/logs.py 
33
34
35
def decompress_log(log: bytes) -> bytes:
    """Decompress a log file"""
    return zlib.decompressobj(wbits=zlib.MAX_WBITS | 16).decompress(log)

human_readable_filesize 

human_readable_filesize(binsize: int) -> str

Convert bytes to human readable string

Source code in blueye/sdk/logs.py 
22
23
24
25
26
27
28
29
30
def human_readable_filesize(binsize: int) -> str:
    """Convert bytes to human readable string"""
    suffix = "B"
    num = binsize
    for unit in ["", "Ki", "Mi"]:
        if abs(num) < 1024.0:
            return f"{num:3.1f} {unit}{suffix}"
        num /= 1024.0
    return f"{num:.1f} Gi{suffix}"