kilowatch-collector.scala

import $ivy.`org.tpolecat::doobie-core:0.4.4`
import $ivy.`org.tpolecat::doobie-postgres:0.4.4`

import java.time._
import java.sql.Timestamp
import java.util.TimeZone
import scala.sys.process._
import scala.xml.XML
import doobie.imports._
import scalaz._, scalaz.effect.IO, Scalaz._, scalaz.concurrent.Task, scalaz.stream.Process

val xa = DriverManagerTransactor[IO]("org.postgresql.Driver", "jdbc:postgresql:kilowatch", "scott", "tiger")

// enable implicit conversion from zoneddatetime
  implicit val ZonedDateTimeMeta: Meta[ZonedDateTime] =
    Meta[Timestamp].nxmap(
      ts  => ZonedDateTime.ofInstant(Instant.ofEpochMilli(ts.getTime), ZoneId.systemDefault), // tz is arbitrary
      zdt => new Timestamp(Instant.from(zdt).toEpochMilli)
    ) 

case class MeterSetting(id: String, joulesPerReading: Int)

case class MeterReading(when: ZonedDateTime, reading: Long, meter: MeterSetting )

def buildCmd (id: String) = s"""/usr/bin/rtlamr -single -filterid=$id -format=xml"""

def read(meter: MeterSetting) = {
  val xml = XML.loadString( buildCmd(meter.id).!! )
  MeterReading( ZonedDateTime.parse( (xml \ "Time") text ), (xml \ "Message" \@ "Consumption") toLong, meter )
}

def buildInsert(reading: MeterReading): Update0 = {
  val reported = java.sql.Timestamp.from(reading.when.toInstant)
  val deviceid = reading.meter.id
  val raw_reading = reading.reading
  val joules = raw_reading * reading.meter.joulesPerReading
  sql"insert into readings (reported, deviceid, raw_reading, joules) values ($reported, $deviceid, $raw_reading, $joules)".update
}

val joulesPerKiloWh = 3600 * 1000;
val joulesPerDecaWh = 3600 * 10;

val house = MeterSetting("ishouldlookthisup", joulesPerKiloWh)
val casita = MeterSetting("thisonetoo", joulesPerDecaWh)

val transaction =
  for {
    h <- buildInsert(read(house)).run
    c <- buildInsert(read(casita)).run
  } yield (h,c)

transaction.transact(xa).unsafePerformIO