redis

Documentation

Index

• type AQIForecast
• type Controller
  • func NewController() (*Controller, error)
  • func (c *Controller) ACKNotifications(ctx context.Context, group string, notifications ...NotificationStream) error
  • func (c *Controller) AddToNotificationStream(ctx context.Context, data ...NotificationStream) error
  • func (c *Controller) CreateConsumerGroup(ctx context.Context, group string) error
  • func (c *Controller) GetAQIFromSensorsInRange(ctx context.Context, longitude, latitude, radius float64) ([]*RawSensorData, error)
  • func (c *Controller) GetAirQuality(ctx context.Context, id int) (*RawSensorData, error)
  • func (c *Controller) GetSensorsInRange(ctx context.Context, longitude, latitude, radius float64) ([]int, error)
  • func (c *Controller) GetTimeSeriesData(ctx context.Context, count int64, ids ...int) (map[UnionKey]*RawQualityData, error)
  • func (c *Controller) NotificationConsumerRead(ctx context.Context, group, consumer string, count int64) ([]NotificationStream, error)
  • func (c *Controller) SetAirQuality(ctx context.Context, data []purpleapi.Response) error
  • func (c *Controller) SetSensorLocationData(ctx context.Context, data []purpleapi.Response) error
  • func (c *Controller) Shutdown() error
• type NotificationStream
• type RawQualityData
• type RawSensorData
• type UnionKey
  • func (u UnionKey) ID() int
  • func (u UnionKey) Timestamp() int

Types

type AQIForecast

type AQIForecast int

AQIForecast indicates changes in the direction of AQI values. In other words it indicates whether or not the AQI is increasing, decreasing, or remaining the same.

const (
	// AQIStatic indicates that the AQI is not changing.
	AQIStatic AQIForecast = iota
	// AQIIncreasing indicates that the AQI is increasing.
	AQIIncreasing
	// AQIDecreasing indicates that the AQI is decreasing.
	AQIDecreasing
)

type Controller

type Controller struct {
	// contains filtered or unexported fields
}

Controller is a container for a Redis client.

func NewController

func NewController() (*Controller, error)

NewController creates a new Redis client.

func (*Controller) ACKNotifications

func (c *Controller) ACKNotifications(ctx context.Context, group string, notifications ...NotificationStream) error

ACKNotifications acknowledges that a set of notifications have been processed.

func (*Controller) AddToNotificationStream

func (c *Controller) AddToNotificationStream(ctx context.Context, data ...NotificationStream) error

AddToNotificationStream adds one or more NotifcationStream items into the forecast stream.

func (*Controller) CreateConsumerGroup

func (c *Controller) CreateConsumerGroup(ctx context.Context, group string) error

CreateConsumerGroup creates a consumer group and the associated stream.

func (*Controller) GetAQIFromSensorsInRange

func (c *Controller) GetAQIFromSensorsInRange(ctx context.Context, longitude, latitude, radius float64) ([]*RawSensorData, error)

GetAQIFromSensorsInRange returns raw sensor data from all sensors within the specified radius around the given coordinates.

func (*Controller) GetAirQuality

func (c *Controller) GetAirQuality(ctx context.Context, id int) (*RawSensorData, error)

GetAirQuality gets 10 most recent PM2.5 AQI readings from a specific sensor.

func (*Controller) GetSensorsInRange

func (c *Controller) GetSensorsInRange(ctx context.Context, longitude, latitude, radius float64) ([]int, error)

GetSensorsInRange takes a pair of coordinates and a radius (in meters) and returns a list of sensor IDs within that circle (if any exist).

func (*Controller) GetTimeSeriesData

func (c *Controller) GetTimeSeriesData(ctx context.Context, count int64, ids ...int) (map[UnionKey]*RawQualityData, error)

GetTimeSeriesData takes a list of sensor IDs and returns the time-series sensor and computed data for each sensor.

func (*Controller) NotificationConsumerRead

func (c *Controller) NotificationConsumerRead(ctx context.Context, group, consumer string, count int64) ([]NotificationStream, error)

NotificationConsumerRead reads n notifications from the "notifications" stream and serializes them into an array of NotificationStream structs.

func (*Controller) SetAirQuality

func (c *Controller) SetAirQuality(ctx context.Context, data []purpleapi.Response) error

SetAirQuality takes an array of Purple Air API response structs and stores the most recent PM2.5 AQI readings from those sensors.

func (*Controller) SetSensorLocationData

func (c *Controller) SetSensorLocationData(ctx context.Context, data []purpleapi.Response) error

SetSensorLocationData takes an array of Purple Air API response structs and creates a map of all sensors in the network.

func (*Controller) Shutdown

func (c *Controller) Shutdown() error

Shutdown closes the connection to the Redis datastore.

type NotificationStream

type NotificationStream struct {
	MessageID string
	UID       int
	AQI       float64
	Forecast  AQIForecast
}

NotificationStream contains data to insert into Redis stream that contains changing AQI information.

type RawQualityData

type RawQualityData struct {
	Time int     `json:"time"`
	PM25 float64 `json:"pm25"`
	AQI  float64 `json:"aqi,omitempty"`
}

RawQualityData contains a time stamp the corresponding pm2.5 measurement.

type RawSensorData

type RawSensorData struct {
	ID   int               `json:"sensor_id"`
	Data []*RawQualityData `json:"measurements"`
}

RawSensorData contains raw sensor from the Redis datastore.

type UnionKey

type UnionKey [2]int

UnionKey is a tuple of an ID and a timestamp. Compared to structs as keys in maps, using a fixed-size array is about 1.5x faster.

func (UnionKey) ID

func (u UnionKey) ID() int

ID returns the ID field from the union.

func (UnionKey) Timestamp

func (u UnionKey) Timestamp() int

Timestamp returns the timestamp field from the union.