Skip to content

Acrclient

module containing the ACRCloud client.

ACRClient

Bases: Client

ACRCloud client wrapper to fetch metadata.

bearer_token: The bearer token for ACRCloud.
Source code in suisa_sendemeldung/acrclient.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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
class ACRClient(Client):
    """ACRCloud client wrapper to fetch metadata.

    Arguments:
    ---------
        bearer_token: The bearer token for ACRCloud.

    """

    # format of timestamp in api answer
    TS_FMT = "%Y-%m-%d %H:%M:%S"
    # timezone of ACRCloud
    ACR_TIMEZONE = "UTC"

    def __init__(
        self: Self, bearer_token: str, base_url: str = "https://eu-api-v2.acrcloud.com"
    ) -> None:
        """Init subclass with default_date."""
        super().__init__(bearer_token=bearer_token, base_url=base_url)
        self.default_date: date = date.today() - timedelta(days=1)  # noqa: DTZ011

    def get_data(
        self: Self,
        project_id: int,
        stream_id: str,
        requested_date: date | None = None,
        timezone: str = ACR_TIMEZONE,
    ) -> Any:  # noqa: ANN401
        """Fetch metadata from ACRCloud for `stream_id`.

        Arguments:
        ---------
            project_id: The Project ID of the stream.
            stream_id: The ID of the stream.
            requested_date: The date of the entries you want (default: yesterday).
            timezone: The timezone to use for localization.

        Returns:
        -------
            json: The ACR data from date

        """
        if requested_date is None:
            requested_date = self.default_date
        data = self.get_bm_cs_projects_results(
            project_id=project_id,
            stream_id=stream_id,
            params=GetBmCsProjectsResultsParams(
                type="day",
                date=requested_date.strftime("%Y%m%d"),
            ),
        )
        for entry in data:
            metadata = entry.get("metadata")
            ts_utc = pytz.utc.localize(
                datetime.strptime(metadata.get("timestamp_utc"), ACRClient.TS_FMT),  # noqa: DTZ007
            )
            ts_local = ts_utc.astimezone(pytz.timezone(timezone))
            metadata.update({"timestamp_local": ts_local.strftime(ACRClient.TS_FMT)})

        return data

    def get_interval_data(  # noqa: ANN201
        self: Self,
        project_id: int,
        stream_id: str,
        start: date,
        end: date,
        timezone: str = ACR_TIMEZONE,
    ):  # pylint: disable-msg=too-many-locals,too-many-arguments
        """Get data specified by interval from start to end.

        Arguments:
        ---------
            project_id: The ID of the project.
            stream_id: The ID of the stream.
            start: The start date of the interval.
            end: The end date of the interval.
            timezone (optional): will be passed to `get_data()`.

        Returns:
        -------
            json: The ACR data from start to end.

        """
        trim = False
        # if we have to localize the timestamps we may need more data
        if timezone != ACRClient.ACR_TIMEZONE:
            # compute utc offset
            offset = pytz.timezone(timezone).utcoffset(datetime.now())  # noqa: DTZ005
            # decrease start by 1 day if we're ahead of utc
            if offset > timedelta(seconds=1):
                computed_start = start - timedelta(days=1)
                computed_end = end
                trim = True
            # increase end by 1 day if we're behind of utc
            elif offset < timedelta(seconds=1):
                computed_start = start
                computed_end = end + timedelta(days=1)
                trim = True
        else:
            computed_start = start
            computed_end = end

        dates = []
        ptr = computed_start
        while ptr <= computed_end:
            dates.append(ptr)
            ptr += timedelta(days=1)
        data = []
        # make the prefix longer by this amount so tqdm lines up with
        # the one in the main code
        ljust_amount: int = 27
        for ptr in tqdm(dates, desc="load ACRCloud data".ljust(ljust_amount)):
            data += self.get_data(
                project_id,
                stream_id,
                requested_date=ptr,
                timezone=timezone,
            )

        # if timestamps are localized we will have to removed the unneeded entries.
        if trim:
            for entry in reversed(data):
                metadata = entry.get("metadata")
                timestamp = metadata.get("timestamp_local")
                timestamp_date = datetime.strptime(timestamp, ACRClient.TS_FMT).date()  # noqa: DTZ007
                if timestamp_date < start or timestamp_date > end:
                    data.remove(entry)

        return data

__init__(bearer_token, base_url='https://eu-api-v2.acrcloud.com')

Init subclass with default_date.

Source code in suisa_sendemeldung/acrclient.py 
28
29
30
31
32
33
def __init__(
    self: Self, bearer_token: str, base_url: str = "https://eu-api-v2.acrcloud.com"
) -> None:
    """Init subclass with default_date."""
    super().__init__(bearer_token=bearer_token, base_url=base_url)
    self.default_date: date = date.today() - timedelta(days=1)  # noqa: DTZ011

get_data(project_id, stream_id, requested_date=None, timezone=ACR_TIMEZONE)

Fetch metadata from ACRCloud for stream_id.

project_id: The Project ID of the stream.
stream_id: The ID of the stream.
requested_date: The date of the entries you want (default: yesterday).
timezone: The timezone to use for localization.

json: The ACR data from date
Source code in suisa_sendemeldung/acrclient.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def get_data(
    self: Self,
    project_id: int,
    stream_id: str,
    requested_date: date | None = None,
    timezone: str = ACR_TIMEZONE,
) -> Any:  # noqa: ANN401
    """Fetch metadata from ACRCloud for `stream_id`.

    Arguments:
    ---------
        project_id: The Project ID of the stream.
        stream_id: The ID of the stream.
        requested_date: The date of the entries you want (default: yesterday).
        timezone: The timezone to use for localization.

    Returns:
    -------
        json: The ACR data from date

    """
    if requested_date is None:
        requested_date = self.default_date
    data = self.get_bm_cs_projects_results(
        project_id=project_id,
        stream_id=stream_id,
        params=GetBmCsProjectsResultsParams(
            type="day",
            date=requested_date.strftime("%Y%m%d"),
        ),
    )
    for entry in data:
        metadata = entry.get("metadata")
        ts_utc = pytz.utc.localize(
            datetime.strptime(metadata.get("timestamp_utc"), ACRClient.TS_FMT),  # noqa: DTZ007
        )
        ts_local = ts_utc.astimezone(pytz.timezone(timezone))
        metadata.update({"timestamp_local": ts_local.strftime(ACRClient.TS_FMT)})

    return data

get_interval_data(project_id, stream_id, start, end, timezone=ACR_TIMEZONE)

Get data specified by interval from start to end.

project_id: The ID of the project.
stream_id: The ID of the stream.
start: The start date of the interval.
end: The end date of the interval.
timezone (optional): will be passed to `get_data()`.

json: The ACR data from start to end.
Source code in suisa_sendemeldung/acrclient.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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
def get_interval_data(  # noqa: ANN201
    self: Self,
    project_id: int,
    stream_id: str,
    start: date,
    end: date,
    timezone: str = ACR_TIMEZONE,
):  # pylint: disable-msg=too-many-locals,too-many-arguments
    """Get data specified by interval from start to end.

    Arguments:
    ---------
        project_id: The ID of the project.
        stream_id: The ID of the stream.
        start: The start date of the interval.
        end: The end date of the interval.
        timezone (optional): will be passed to `get_data()`.

    Returns:
    -------
        json: The ACR data from start to end.

    """
    trim = False
    # if we have to localize the timestamps we may need more data
    if timezone != ACRClient.ACR_TIMEZONE:
        # compute utc offset
        offset = pytz.timezone(timezone).utcoffset(datetime.now())  # noqa: DTZ005
        # decrease start by 1 day if we're ahead of utc
        if offset > timedelta(seconds=1):
            computed_start = start - timedelta(days=1)
            computed_end = end
            trim = True
        # increase end by 1 day if we're behind of utc
        elif offset < timedelta(seconds=1):
            computed_start = start
            computed_end = end + timedelta(days=1)
            trim = True
    else:
        computed_start = start
        computed_end = end

    dates = []
    ptr = computed_start
    while ptr <= computed_end:
        dates.append(ptr)
        ptr += timedelta(days=1)
    data = []
    # make the prefix longer by this amount so tqdm lines up with
    # the one in the main code
    ljust_amount: int = 27
    for ptr in tqdm(dates, desc="load ACRCloud data".ljust(ljust_amount)):
        data += self.get_data(
            project_id,
            stream_id,
            requested_date=ptr,
            timezone=timezone,
        )

    # if timestamps are localized we will have to removed the unneeded entries.
    if trim:
        for entry in reversed(data):
            metadata = entry.get("metadata")
            timestamp = metadata.get("timestamp_local")
            timestamp_date = datetime.strptime(timestamp, ACRClient.TS_FMT).date()  # noqa: DTZ007
            if timestamp_date < start or timestamp_date > end:
                data.remove(entry)

    return data