From 1cd3e59e31df2c8ba7ad1d9274c13f8dc94e58d5 Mon Sep 17 00:00:00 2001 From: Troy Raen Date: Sun, 30 Jun 2024 16:30:40 -0700 Subject: [PATCH] import full modules for clarity --- pittgoogle/bigquery.py | 24 ++++++++++++++---------- pittgoogle/utils.py | 24 ++++++++++++------------ 2 files changed, 26 insertions(+), 22 deletions(-) diff --git a/pittgoogle/bigquery.py b/pittgoogle/bigquery.py index 0d9bb16..668c370 100644 --- a/pittgoogle/bigquery.py +++ b/pittgoogle/bigquery.py @@ -2,9 +2,8 @@ """Classes to facilitate connections to BigQuery datasets and tables.""" import logging +import attrs import google.cloud.bigquery -from attrs import define, field -from attrs.validators import instance_of, optional from .alert import Alert from .auth import Auth @@ -12,7 +11,7 @@ LOGGER = logging.getLogger(__name__) -@define +@attrs.define class Table: """Methods and properties for a BigQuery table. @@ -33,17 +32,22 @@ class Table: """ # Strings _below_ the field will make these also show up as individual properties in rendered docs. - name: str = field() + name: str = attrs.field() """Name of the BigQuery table.""" - dataset: str = field() + dataset: str = attrs.field() """Name of the BigQuery dataset this table belongs to.""" # The rest don't need string descriptions because they are explicitly defined as properties below. - _projectid: str = field(default=None) - _auth: Auth = field(default=None, validator=optional(instance_of(Auth))) - _client: google.cloud.bigquery.Client | None = field( - default=None, validator=optional(instance_of(google.cloud.bigquery.Client)) + _projectid: str = attrs.field(default=None) + _auth: Auth = attrs.field( + default=None, validator=attrs.validators.optional(attrs.validators.instance_of(Auth)) ) - _table: google.cloud.bigquery.Table | None = field(default=None, init=False) + _client: google.cloud.bigquery.Client | None = attrs.field( + default=None, + validator=attrs.validators.optional( + attrs.validators.instance_of(google.cloud.bigquery.Client) + ), + ) + _table: google.cloud.bigquery.Table | None = attrs.field(default=None, init=False) @classmethod def from_cloud( diff --git a/pittgoogle/utils.py b/pittgoogle/utils.py index b4e9467..e4e404d 100644 --- a/pittgoogle/utils.py +++ b/pittgoogle/utils.py @@ -2,19 +2,19 @@ """Classes and functions to support working with alerts and related data.""" import json import logging -from base64 import b64decode, b64encode +import base64 from collections import OrderedDict from io import BytesIO +import astropy.table +import astropy.time +import attrs import fastavro -from astropy.table import Table -from astropy.time import Time -from attrs import define LOGGER = logging.getLogger(__name__) -@define +@attrs.define class Cast: """Methods to convert data types.""" @@ -33,7 +33,7 @@ def bytes_to_b64utf8(bytes_data): ``bytes_data`` converted to a string in UTF-8 format """ if bytes_data is not None: - return b64encode(bytes_data).decode("utf-8") + return base64.b64encode(bytes_data).decode("utf-8") @staticmethod def json_to_dict(bytes_data): @@ -65,7 +65,7 @@ def b64json_to_dict(bytes_data): ``bytes_data`` unpacked into a dictionary. """ if bytes_data is not None: - return Cast.json_to_dict(b64decode(bytes_data)) + return Cast.json_to_dict(base64.b64decode(bytes_data)) @staticmethod def avro_to_dict(bytes_data): @@ -102,11 +102,11 @@ def b64avro_to_dict(bytes_data): data : `dict` ``bytes_data`` unpacked into a dictionary. """ - return Cast.avro_to_dict(b64decode(bytes_data)) + return Cast.avro_to_dict(base64.b64decode(bytes_data)) # --- Work with alert dictionaries @staticmethod - def alert_dict_to_table(alert_dict: dict) -> Table: + def alert_dict_to_table(alert_dict: dict) -> astropy.table.Table: """Package a ZTF alert dictionary into an Astopy Table.""" # collect rows for the table candidate = OrderedDict(alert_dict["candidate"]) @@ -117,7 +117,7 @@ def alert_dict_to_table(alert_dict: dict) -> Table: rows.append(prv_cand_tmp) # create and return the table - table = Table(rows=rows) + table = astropy.table.Table(rows=rows) table.meta["comments"] = f"ZTF objectId: {alert_dict['objectId']}" return table @@ -136,7 +136,7 @@ def _strip_cutouts_ztf(alert_dict: dict) -> dict: # dates @staticmethod - def jd_to_readable_date(jd): + def jd_to_readable_date(jd) -> str: """Convert a Julian date to a human readable string. Parameters @@ -149,7 +149,7 @@ def jd_to_readable_date(jd): date : `str` ``jd`` in the format 'day mon year hour:min' """ - return Time(jd, format="jd").strftime("%d %b %Y - %H:%M:%S") + return astropy.time.Time(jd, format="jd").strftime("%d %b %Y - %H:%M:%S") # --- Survey-specific