Skip to content

Commit f2e67ad

Browse files
VallascVallasc
committed
Code clean
1 parent 77d5516 commit f2e67ad

File tree

2 files changed

+136
-83
lines changed

2 files changed

+136
-83
lines changed

‎src/main/scala/Main.scala‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
import _root_.Utils.FileUtils
2+
import java.nio.charset.Charset
3+
import breeze.linalg.DenseMatrix
4+
import Algorithms.Invert
5+
import Algorithms.MedianFilter
6+
import Algorithms.MeanFilter
7+
import Pipelines.{EdgeDetection, MedianDenoiser, GibbsEdgeDetection, Pipeline}
8+
import Algorithms.Denoiser
9+
import Pipelines.GibbsDenoiser
10+
11+
object Main {
12+
13+
var inputPathImage = "./data/input.png"
14+
var outputPathImage = "./data/out.png"
15+
var outputPathJson = "./data/report.json"
16+
var inputPiepeline = "GibbsEdgeDetection"
17+
18+
var padding = 3
19+
var subHeight = 100
20+
var subWidth = 100
21+
var denoiserRuns = 90
22+
var debug = 1
23+
val usage = "\nUsage: [--sub_matrix_size] [--padding] [--denoiser_runs] [--pipeline] [--debug] [--output_file_json] [--output_file_image] input_file_image\n"
24+
25+
def main(args: Array[String]): Unit = {
26+
// Check arguments
27+
if (args.length % 2 == 0) {
28+
println(usage)
29+
return
30+
}
31+
args.sliding(2, 2).toList.collect {
32+
case Array("--sub_matrix_size", m_size: String) => {
33+
subHeight = m_size.toInt
34+
subWidth = m_size.toInt
35+
}
36+
case Array("--padding", p: String) => padding = p.toInt
37+
case Array("--denoiser_runs", runs: String) => denoiserRuns = runs.toInt
38+
case Array("--debug", d: String) => debug = d.toInt
39+
case Array("--output_file_json", out: String) => outputPathJson = out
40+
case Array("--output_file_image", out: String) => outputPathImage = out
41+
case Array("--pipeline", out: String) => inputPiepeline = out
42+
case Array(out: String) => inputPathImage = out
43+
}
44+
45+
println(s"\nInput file: ${inputPathImage}")
46+
println(s"Output file: ${outputPathImage}")
47+
println(s"Output json: ${outputPathJson}")
48+
println(s"Sub matrix size: ${subHeight}")
49+
println(s"Paddding: ${padding}")
50+
51+
val inputStream = FileUtils.getInputStream(inputPathImage)
52+
val inputImage = new Image()
53+
val pixelArray = inputImage.getPixelMatrix(inputStream, true)
54+
inputStream.close()
55+
val pixelMatrix = new DenseMatrix[Double](inputImage.width, inputImage.height, pixelArray.map(_.toDouble))
56+
57+
// Define the Spark Job
58+
val job = new SparkJob(padding, subHeight, subWidth, denoiserRuns, debug)
59+
val pipeline = inputPiepeline match {
60+
case "GibbsDenoise" => new GibbsDenoiser(denoiserRuns)
61+
case "GibbsEdgeDetection" => new GibbsEdgeDetection(denoiserRuns)
62+
case "EdgeDetection" => EdgeDetection
63+
case "MedianDenoiser" => MedianDenoiser
64+
}
65+
66+
println("\nStart")
67+
val result = Utils.time(job.run(pixelMatrix, pipeline))
68+
if(debug > 0)
69+
println(s"Time: ${result._2} ms")
70+
71+
val outputStream = FileUtils.getOutputStream(outputPathImage)
72+
val outputImage = new Image()
73+
outputImage.setPixelMatrix(result._1.data.map(_.toInt), result._1.rows, result._1.cols, true)
74+
outputImage.saveImage(outputStream)
75+
outputStream.close()
76+
77+
val json = s"""{
78+
"time": ${result._2},
79+
"inputPiepeline": "${inputPiepeline}",
80+
"inputPathImage": "${inputPathImage}",
81+
"outputPathImage": "${outputPathImage}",
82+
"outputPathJson": "${outputPathJson}",
83+
"denoiserRuns": ${denoiserRuns},
84+
"padding": ${padding},
85+
"subHeight": ${subHeight},
86+
"subWidth": ${subWidth}
87+
}"""
88+
89+
val jsonOutputStream = FileUtils.getOutputStream(outputPathJson)
90+
jsonOutputStream.write(json.getBytes(Charset.forName("UTF-8")))
91+
jsonOutputStream.close()
92+
}
93+
}
94+
95+
// sbt assembly
96+
// spark-submit --class SparkJob ./jar/binary.jar ./data/nike_noisy.png

‎src/main/scala/SparkJob.scala‎

Lines changed: 40 additions & 83 deletions
Original file line numberDiff line numberDiff line change
@@ -10,95 +10,51 @@ import org.apache.spark.ml.image.ImageSchema._
1010
import java.io.PrintWriter
1111
import _root_.Utils.FileUtils
1212
import java.nio.charset.Charset
13-
14-
object SparkJob extends Job {
15-
var inputPathImage = "file:///C://data/img_noisy1.png"
16-
var outputPathImage = "file:///C://data/out.png"
17-
var outputPathJson = "file:///C://data/report.json"
18-
19-
var padding = 10
20-
var subHeight = 300
21-
var subWidth = 300
22-
var denoiserRuns = 100
23-
24-
var debug = 1
25-
26-
val usage = "\nUsage: [--sub_matrix_size] [--padding] [--denoiser_runs] [--debug] [--output_file_json] [--output_file_image] input_file_image\n"
27-
def main(args: Array[String]): Unit = {
28-
// Check arguments
29-
if (args.length % 2 == 0) {
30-
println(usage)
31-
return
32-
}
33-
args.sliding(2, 2).toList.collect {
34-
case Array("--sub_matrix_size", m_size: String) => {
35-
subHeight = m_size.toInt
36-
subWidth = m_size.toInt
37-
}
38-
case Array("--padding", p: String) => padding = p.toInt
39-
case Array("--denoiser_runs", runs: String) => denoiserRuns = runs.toInt
40-
case Array("--debug", d: String) => debug = d.toInt
41-
case Array("--output_file_json", out: String) => outputPathJson = out
42-
case Array("--output_file_image", out: String) => outputPathImage = out
43-
case Array(out: String) => inputPathImage = out
44-
}
45-
46-
println(s"\nInput file: ${inputPathImage}")
47-
println(s"Output file: ${outputPathImage}")
48-
println(s"Output json: ${outputPathJson}")
49-
println(s"Sub matrix size: ${subHeight}")
50-
println(s"Paddding: ${padding}")
51-
52-
println("\nStart")
53-
val t = Utils.time(run)
54-
if(debug > 0)
55-
println(s"Time: $t ms")
56-
57-
val json = "{\"time\":" + t +"}"
58-
val os = FileUtils.getOutputStream(outputPathJson)
59-
os.write(json.getBytes(Charset.forName("UTF-8")))
60-
os.close()
61-
62-
}
63-
64-
def run(): Unit = {
13+
import java.io.InputStream
14+
import java.io.OutputStream
15+
import Pipelines.Pipeline
16+
import scala.collection.parallel.immutable.ParSeq
17+
import org.apache.spark.HashPartitioner
18+
import org.apache.spark.storage.StorageLevel
19+
20+
class SparkJob(val padding: Int = 3,
21+
val subHeight: Int = 100,
22+
val subWidth: Int = 100,
23+
val denoiserRuns: Int = 80,
24+
val debug: Int = 1) extends Serializable {
25+
26+
/**
27+
* Runs the Job on the spark cluster given a Pipeline
28+
*/
29+
def run(inputMatrix: BDM[Double], pipeline: Pipeline): BDM[Double] = {
30+
// Spark setup
6531
val conf = new SparkConf().setAppName("GibbsDenoiser")
6632
//.setMaster("local[*]")
6733
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
68-
conf.registerKryoClasses(Array(classOf[Tuple2[Tuple2[Int, Int], Matrix]]))
69-
34+
7035
val sc = new SparkContext(conf)
7136

72-
val inputImage = new Image()
73-
val is = FileUtils.getInputStream(inputPathImage)
74-
val pixelArray = inputImage.getPixelMatrix(is, true)
75-
is.close()
76-
val pixelMatrix = new BDM[Double](inputImage.width, inputImage.height, pixelArray.map(_.toDouble))
37+
// Split big matrix into submatrixes, according to subHeight and subWidth
38+
val splitted = splitImage(inputMatrix)
7739

78-
val splitted = splitImage(pixelMatrix)
79-
80-
var n = (pixelMatrix.cols) / subWidth // cols divisions
81-
var m = (pixelMatrix.rows) / subHeight // rows divisions
82-
83-
val mat = sc.parallelize(splitted, n * m * 100)
84-
val computed = compute(mat, processPipelne)
40+
// Make RDD of matrixes
41+
//val matrixes = sc.parallelize(splitted._1, splitted._2 * splitted._3 * 100)
42+
val matrixes = sc.parallelize(splitted._1)
43+
matrixes.partitionBy(new HashPartitioner(splitted._2 * splitted._3)).persist(StorageLevel.MEMORY_ONLY)
44+
val computed = compute(matrixes, pipeline)
8545

46+
// Reassemble the matrix
8647
val blockMat = new BlockMatrix(computed, subHeight, subWidth)
8748
val out = Utils.matrixAsBreeze(blockMat.toLocalMatrix())
88-
val cleaned = out(0 to pixelMatrix.rows -1, 0 to pixelMatrix.cols -1).copy
89-
println("It's all ok")
9049

91-
val os = FileUtils.getOutputStream(outputPathImage)
92-
val outputImage = new Image()
93-
outputImage.setPixelMatrix(cleaned.data.map(_.toInt), cleaned.rows, cleaned.cols, true)
94-
outputImage.saveImage(os)
95-
os.close()
50+
// Remove padding border
51+
out(0 to inputMatrix.rows -1, 0 to inputMatrix.cols -1).copy
9652

9753
//edges.partitionBy(new RangePartitioner(SparkContextSingleton.DEFAULT_PARALLELISM, edges)).persist(StorageLevel.MEMORY_AND_DISK)
9854
}
9955

10056

101-
private def splitImage(pixelMatrix: BDM[Double]): Seq[((Int, Int), Matrix)] = {
57+
private def splitImage(pixelMatrix: BDM[Double]): (Seq[((Int, Int), Matrix)], Int, Int) = {
10258
val subHeight = if(this.subHeight <= 0) pixelMatrix.rows else this.subHeight
10359
val subWidth = if(this.subWidth <= 0) pixelMatrix.cols else this.subWidth
10460
assert(padding <= subHeight)
@@ -131,27 +87,28 @@ object SparkJob extends Job {
13187
println("x sub-matrix: " + n)
13288
println("y sub-matrix: " + m)
13389
}
134-
for {
135-
p1 <- 0 until n // X
136-
p2 <- 0 until m // Y
90+
(for {
91+
p1 <- (0 until n) // X
92+
p2 <- (0 until m) // Y
13793
} yield {
13894
val xFromPadded = p1 * subWidth
13995
val xToPadded = xFromPadded + subWidth + padding*2 -1
14096
val yFromPadded = p2 * subHeight
14197
val yToPadded = yFromPadded + subHeight + padding*2 -1
14298
val matrix = paddedMatrix(yFromPadded to yToPadded, xFromPadded to xToPadded).copy
14399
((p2, p1), Utils.matrixFromBreeze(matrix))
144-
}
100+
}, n, m)
145101
}
146102

147-
private def compute(matrixes : RDD[((Int, Int), Matrix)], transform: (BDM[Double]) => (BDM[Double])): RDD[((Int, Int), Matrix)] = {
103+
private def compute(matrixes : RDD[((Int, Int), Matrix)], pipeline: Pipeline): RDD[((Int, Int), Matrix)] = {
104+
// mapPartitions??
148105
matrixes.map ( element => {
149106
val matrix = Utils.matrixAsBreeze(element._2)
150-
val out = transform(matrix)
107+
val out = removePadding(pipeline.run(matrix))
151108
(element._1, Utils.matrixFromBreeze(out))
152109
})
153110
}
154111

155-
}
156-
// sbt "runMain SparkJob ./data/nike_noisy.png"
157-
// spark-submit --class SparkJob ./jar/binary.jar ./data/nike_noisy.png
112+
def removePadding(matrix: BDM[Double]) : BDM[Double] =
113+
matrix(padding to -padding, padding to -padding).copy
114+
}

0 commit comments

Comments
 (0)