Source code for onbasca.base.models.relaybw

# SPDX-FileCopyrightText: 2022 The Tor Project, Inc.
#
# SPDX-License-Identifier: BSD-3-Clause

import datetime
import logging

from django.db import models

from .base import BaseModel

logger = logging.getLogger(__name__)


BWLINE_KEYS = [
    "node_id",
    "nick",
    "bw",
    "bw_mean",
    "bw_median",
    "consensus_bandwidth",
    "consensus_bandwidth_is_unmeasured",
    "desc_bw_avg",
    "desc_bw_bur",
    "desc_bw_obs_last",
    "desc_bw_obs_mean",
    "error_circ",
    "error_destination",
    "error_misc",
    "error_second_relay",
    "error_stream",
    "master_key_ed25519",
    "relay_in_recent_consensus_count",
    "relay_recent_measurement_attempt_count",
    "relay_recent_measurements_excluded_error_count",
    "relay_recent_priority_list_count",
    "success",
    "time",
    "measured_at",
    "updated_at",
    "unmeasured",
    "vote",
    "circ_fail",
    "pid_delta",
    "pid_bw",
    "pid_error",
    "pid_error_sum",
]
BWLINE2RELABW = {
    "node_id": "fingerprint",
    "nick": "nickname",
    # "time": "measured_at",
    # "measured_at": "updated_at"
}




class RelayBwManagerBase(models.Manager):
    pass





class RelayBwBase(BaseModel):


    class Meta:
        abstract = True
        unique_together = ["fingerprint", "bwfile"]
        get_latest_by = "measured_at"


    objects = RelayBwManagerBase()

    fingerprint = models.CharField(max_length=40)
    measured_at = models.DateTimeField(editable=True, null=True, blank=True)
    updated_at = models.DateTimeField(editable=True, null=True, blank=True)

    nickname = models.CharField(max_length=19, null=True, blank=True)
    bw = models.PositiveIntegerField(null=True, blank=True)
    consensus_bandwidth = models.PositiveIntegerField(null=True, blank=True)
    consensus_bandwidth_is_unmeasured = models.BooleanField(
        null=True, blank=True
    )
    desc_bw_avg = models.PositiveIntegerField(null=True, blank=True)
    desc_bw_bur = models.PositiveIntegerField(null=True, blank=True)
    desc_bw_obs_last = models.PositiveIntegerField(null=True, blank=True)
    error_circ = models.PositiveSmallIntegerField(null=True, blank=True)
    error_stream = models.PositiveSmallIntegerField(null=True, blank=True)
    relay_in_recent_consensus_count = models.PositiveSmallIntegerField(
        null=True, blank=True
    )
    relay_recent_measurement_attempt_count = models.PositiveSmallIntegerField(
        null=True, blank=True
    )
    relay_recent_measurement_failure_count = models.PositiveSmallIntegerField(
        null=True, blank=True
    )
    relay_recent_measurements_excluded_error_count = (
        models.PositiveSmallIntegerField(null=True, blank=True)
    )
    relay_recent_priority_list_count = models.PositiveSmallIntegerField(
        null=True, blank=True
    )
    success = models.PositiveSmallIntegerField(null=True, blank=True)
    vote = models.BooleanField(null=True, blank=True)
    under_min_report = models.BooleanField(null=True, blank=True)

    def __str__(self):
        return self.nickname

    # for admin


    def relay_relaydesc_latest(self):
        if not self.relay:
            return None
        return self.relay.relaydesc_latest()




    def relay_routerstatus_latest(self):
        if not self.relay:
            return None
        return self.relay.routerstatus_latest()




    def is_exit(self):
        rs_last = self.relay_routerstatus_latest()
        if rs_last:
            return rs_last.is_exit
        return None




    def to_str_v15(self):
        from onbasca.base import constants

        kwargs = dict([(k, getattr(self, k, "")) for k in BWLINE_KEYS])
        for k, v in BWLINE2RELABW.items():
            kwargs[k] = getattr(self, v, None)
        kwargs["node_id"] = "$" + kwargs["node_id"]
        kwargs.pop("measured_at")
        kwargs.pop("updated_at")
        if kwargs["vote"] is not False:
            kwargs.pop("vote")
        if not kwargs["unmeasured"]:
            kwargs.pop("unmeasured")

        key_values = []
        for key, value in sorted(kwargs.items()):
            logger.debug("%s: %s", key, value)
            # Ignore Torflow KeyValues atm.
            if key.startswith("pid") or key.startswith("circ"):
                continue
            if isinstance(value, bool):
                value = int(value)
            if value is None:
                value = 0
            if isinstance(value, datetime.datetime):
                value = value.strftime(constants.DATETIME_FORMAT)

            key_values.append("{}={}".format(key, value))
        key_values_str = " ".join(key_values)
        return key_values_str




    def to_str(self):
        return self.to_str_v15()