AkkaClient.scala

import akka.actor.ActorSystem
import akka.stream.scaladsl.Sink
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.model._
import akka.http.scaladsl.Http
import HttpMethods._
import akka.stream.ActorMaterializer
import Uri.Query
import scala.concurrent.Future
import spray.json._
import scala.util.{Success, Failure}

case class Sensor(name: String, data: SensorData)
case class SensorData(averageSpeed: AverageSpeed)
case class AverageSpeed(meta: Meta, data: Map[String, Double])
case class Meta(units: String, theme: String, name: String)

/**
 * Marshalling from JSON and to JSON for Urban Observatory Traffic Sensors
 */
trait Protocols extends SprayJsonSupport with DefaultJsonProtocol {
  implicit val metaFormat: RootJsonFormat[Meta] = 
    jsonFormat(Meta.apply, "units", "theme", "name")
  implicit val averagespeedFormat: RootJsonFormat[AverageSpeed] = 
    jsonFormat(AverageSpeed.apply, "meta", "data")
  implicit val sensorDataFormat: RootJsonFormat[SensorData] = 
    jsonFormat(SensorData.apply, "Average Speed")
  implicit val sensorFormat: RootJsonFormat[Sensor] = 
    jsonFormat(Sensor.apply, "name", "data")
}

object TestJson extends App with Protocols {
  val json_string = scala.io.Source.fromFile("data/traffic_sensor.json").getLines.mkString

  json_string.
    parseJson.
    convertTo[List[Sensor]].
    foreach(println)
}

object Client extends App with Protocols {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  implicit val executionContext = system.dispatcher

  val uri = Uri("http://uoweb1.ncl.ac.uk/api/v1/sensor/data/raw.json")
  val api_key =  // your api key here

  val query: Query = Query("api_key" -> api_key,
    "sensor_name" -> "N05171T",
    "start_time" -> "20170201",
    "end_time" -> "20170202",
    "variable" -> "average speed")

  val res: Future[HttpResponse] = 
    Http().singleRequest(HttpRequest(GET, uri = uri.withQuery(query)))

  res andThen {
    case Success(response) => {
      response.
        entity.
        dataBytes.
        map(_.utf8String).
        map(_.parseJson.convertTo[List[Sensor]]).
        runForeach(println)
    }
    case Failure(ex) => println(ex)
  } onComplete {
    _ => system.terminate()
  }
}