Commits

Scott Bennett-McLeish committed fafcb21

Added code for RabbitMQ overview.

  • Participants
  • Parent commits ecef6c2

Comments (0)

Files changed (5)

File fitbit_weight/FitBitCollector.groovy

+
+//String sourceText = new URL("http://www.fitbit.com/user/24KH6K/weight/date/2013-" +args[0]).text
+String sourceText = new URL("http://www.fitbit.com/user/24KH6K/weight").text
+
+//working
+String weight = sourceText.findAll(/[\s]+?([0-9][0-9][0-9]?\.[0-9]?)(?=\skg)/)[0]
+println "weight: ${weight.trim()}"
+
+//not quite working
+println "bmi:"
+//sourceText.findAll(/(?!=id\=\"bmi\".+?>)[\s]+?([0-9][0-9][0-9]?\.?[0-9]?)[\s][\s](?=[\s]?.+</span>.+?</div>)/).each { println it }
+def bmi = sourceText.findAll(/(?!=id\\=\\"bmi\\".+?>)[\\s]+?([0-9][0-9][0-9]?\\.?[0-9]?)[\\s][\\s](?=[\\s]?.+<\/span>.+?<\/div>)/)
+println bmi
+//.each { println it }

File rabbitmq_messaging/ExtractTemperatures.groovy

+import groovy.json.*
+
+class ExtractTemperatures {
+	
+	def processBOM(Expando site) {
+		//create a URL and get the text from it
+		def content = new URL(site.source).text
+
+		//create a new jsonSluper and parse the json data
+		def jsonSlurper = new JsonSlurper()
+		def jsonData = jsonSlurper.parseText(content)
+		
+		//extract just the entry we want
+		def latestEntry = jsonData.observations.data[0]
+		
+		println "\n==== BOM ${site.name} ===="
+		println "airTemp = ${latestEntry.air_temp}"
+		println "windSpeed = ${latestEntry.wind_spd_kmh}"
+		println "windDirection = ${latestEntry.wind_dir}"
+		
+		return latestEntry.air_temp
+	}
+	
+	def processAgDeptWithRegex(Expando site) {
+		//generic URL values
+		String agDeptPrefix = "http://agspsrv34.agric.wa.gov.au/climate/livedata/"
+		String agDeptsuffix = "webpag.htm"
+		
+		//retrieve web page text for the particular site
+		String sourceText = new URL(agDeptPrefix + site.source + agDeptsuffix).text
+		
+		//extract all lines with the "Yellow" formatting, then just get the first one (which we know is the correct one) for Temperature.
+		String firstYellowRow = sourceText.findAll(/<td width="68"><font face="Courier New"[\s]?color="yellow">[\s]?(?:[0-9]+\.[0-9]?)[<\/font>]?<\/td>/)[0]
+		String currentTemp = firstYellowRow.findAll(/[0-9]+\.[0-9]/)[0] //just get the "decimal" element
+		
+		//extract all lines with the "Yellow" formatting, then just get the sixth one (which we know is the correct one) for Wind Speed.
+		String windSpeedElement = sourceText.findAll(/<td width="68"><font face="Courier New"[\s]?color="yellow">[\s]?(?:[0-9]+\.[0-9]?)[<\/font>]?<\/td>/)[5]
+		String currentWindSpeed = windSpeedElement.findAll(/[0-9]+\.[0-9]/)[0] //just get the "decimal" element
+		
+		//extract all lines with the "Yellow" formatting, then just get the one with ENSW variations in it for Wind Direction.
+		String windDirectionElement = sourceText.findAll(/<td width="68"><font face="Courier New"[\s]?color="yellow">[\s]?(?:[ENSW]+)[\s]?[<\/font>]?<\/td>/)[0]
+		String currentWindDirection = windDirectionElement.findAll(/(?!=ow">[\s]?)[ENSW]+[\s]?(?=<\/)/)[0] //just get the element value
+		
+		println "\n==== Ag. Dept. ${site.name} ===="
+		println "airTemp = ${currentTemp}"
+		println "windSpeed = ${currentWindSpeed}"
+		println "windDirection = ${currentWindDirection}"
+	
+		return currentTemp
+	}
+}

File rabbitmq_messaging/RunWeatherMessager.groovy

+if (args[0] == "pub") {
+	println "Starting Weather producer"
+	def pub = new WeatherProducer()
+	pub.bomSite = new Expando(source: "http://www.bom.gov.au/fwo/IDW60901/IDW60901.94608.json", name: "Perth")
+	pub.start()
+}
+else if (args[0] == "sub") {
+	println "Starting Weather consumer"
+	def sub = new WeatherConsumer()
+	sub.execute()
+}
+else {
+	println "===ERROR==="
+	println "Only valid options are 'pub' or 'sub'."
+}

File rabbitmq_messaging/WeatherConsumer.groovy

+@Grapes(
+	@Grab(group='com.rabbitmq', module='amqp-client', version='3.0.1')
+)
+import com.rabbitmq.client.*
+
+class WeatherConsumer {
+	ConnectionFactory factory = null
+	Connection conn = null
+	String exchangeName = "WeatherExchange"
+	String routingKey = "weather"
+	String queueName = "WeatherQueue"
+	
+	def execute() {
+		Channel channel = getNewChannel()
+		channel.queueBind(queueName, exchangeName, "#");
+		println " Queue: ${queueName} "
+
+		boolean noAck = false;
+		def consumer = new QueueingConsumer(channel);
+		channel.basicConsume(queueName, noAck, consumer);
+		boolean running = true
+
+		while(running) {
+			QueueingConsumer.Delivery delivery;
+			try {
+				delivery = consumer.nextDelivery();
+				println new String(delivery.body) + " - " + System.currentTimeMillis()
+			} catch (InterruptedException ie) {
+				//we don't really care .. do we?
+				println ie.getMessage()
+				running = false
+			}
+			
+			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+		}
+	}
+	
+	Channel getNewChannel() {
+		if (factory == null || conn == null) {
+			factory = new ConnectionFactory()
+			factory.setUsername("guest")
+			factory.setPassword("guest")
+			factory.setVirtualHost("/")
+			factory.setHost("localhost")
+			factory.setPort(5672)
+			conn = factory.newConnection()
+		}
+		
+		return conn.createChannel()
+	}
+}

File rabbitmq_messaging/WeatherProducer.groovy

+@Grapes(
+	@Grab(group='com.rabbitmq', module='amqp-client', version='3.0.1')
+)
+import com.rabbitmq.client.*
+
+class WeatherProducer {
+	ConnectionFactory factory = null
+	Connection conn = null
+	String exchangeName = "WeatherExchange"
+	String routingKey = "weather"
+	String queueName = "WeatherQueue"
+	Expando bomSite = null
+		
+	def start() {
+		Channel channel = getNewChannel()
+		channel.exchangeDeclare(exchangeName, "fanout", true)
+		channel.queueDeclare(queueName, true, false, false, null)
+		def extractTemperatures = new ExtractTemperatures()
+
+		while (true) {
+			int sleepTime = Math.random()*30000
+			
+			String myData = extractTemperatures.processBOM(bomSite)
+			
+			channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, myData.getBytes())
+
+			println "Sleeping for ${sleepTime}ms"
+			sleep(sleepTime)
+		}
+
+		channel.close()
+		conn.close()
+	}
+	
+	Channel getNewChannel() {
+		if (factory == null || conn == null) {
+			factory = new ConnectionFactory()
+			factory.setUsername("guest")
+			factory.setPassword("guest")
+			factory.setVirtualHost("/")
+			factory.setHost("localhost")
+			factory.setPort(5672)
+			conn = factory.newConnection()
+		}
+		
+		return conn.createChannel()
+	}
+}