1
\$\begingroup\$

Came back to the software side of my night vision security camera project after taking some time to learn about electronics and I found it a bit hard to retrace the logic of my program.

I have a bunch of sensors that communicate their state to different workers and I think I fudged the design big time.

Here is a simple graphic of how my gophers go about their business, communicating via channels.

Camera----------------StreamServer-------------Robot------------LightSensor

StreamFrames --------->

LocalVideo   ---------------------------------->
                      BusyOrIdle--------------->
                      <------------------------BusyOrIdle
                      <-----------------------------------------LightLevel
                                               <----------------LightLevel

And the code.

main.go

package main

import (
    "apps/picam/camera"
    "apps/picam/hardware"
    "apps/picam/server"
    "flag"
    "log"
    "os"
    "os/signal"
    "time"

    rpio "github.com/stianeikeland/go-rpio"
)

var light int

func init() {
    if err := rpio.Open(); err != nil {
        log.Fatal(err)
    }

    hardware.PinsPullDown()
    hardware.SetPinMode()
    hardware.SetPinState()
}
func main() {
    var port = flag.String("port", ":5000", "The port on which the camera listens for clients")
    var password = flag.String("pwd", "", "Camera password")

    flag.Parse()

    defer rpio.Close()

    stream := make(chan []byte, 1)    //chan for streaming video to client
    localFeed := make(chan []byte, 1) //chan for local video recording

    robotState := make(chan string, 1)
    serverState := make(chan string, 1)

    l1 := make(chan int, 1)
    l2 := make(chan int, 1)

    done := make(chan bool, 1)

    videoFeedListeners := []chan []byte{stream, localFeed}

    videoPipe, err := camera.Start("/home/pi/src/picam/v4l2")
    checkErr(err)

    defer videoPipe.Close()

    security := hardware.SecurityRobot{
        VideoFeed:   localFeed, //used to record video locally
        Done:        done,
        RobotState:  robotState,  //refers to, if motion sensor detected movement and camera is recording
        ServerState: serverState, //refers to, if server is streaming
    }

    cam := camera.Camera{
        Done:      done,
        Reader:    videoPipe,
        Forwarder: videoFeedListeners,
    }

    server := server.NewServer(robotState, serverState)

    light := hardware.ReadLightLevels(done)
    motion := hardware.DetectMovement(done)

    go cam.ReadForwardFrame()
    go cam.Statistics()

    go func(done chan bool) {
        for {
            select {
            case <-done:
                log.Println("Light sensor communicator exit - √")
                return
            case level := <-light:
                l1 <- level
                l2 <- level
            }
        }
    }(done)

    go server.Run(port, password, stream, l2, done)
    go security.Video(l1, motion)

    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt)
    for range c {
        hardware.PinsPullDown()
        close(done)
        break
    }

    log.Println("Interrupt signal received. Shutting camera down")
    time.Sleep(2 * time.Second)
}

func checkErr(err error) {
    if err != nil {
        log.Fatal(err)
    }
}

camera.go

package camera

import (
    "bytes"
    "fmt"
    "io"
    "log"
    "os/exec"
    "time"
)

type Camera struct {
    Done      chan bool
    Reader    io.ReadCloser
    Forwarder []chan []byte
    FPS       int
    FrameSize int
}

func Start(command string) (io.ReadCloser, error) {
    var stderr bytes.Buffer
    cmd := exec.Command(command)
    cmd.Stderr = &stderr

    pipe, _ := cmd.StdoutPipe()

    if err := cmd.Start(); err != nil {
        return nil, fmt.Errorf(fmt.Sprint(err) + ": " + stderr.String())
    }

    return pipe, nil
}

func (c *Camera) ReadForwardFrame() {
    for {
        frame := make([]byte, 65507)

        n, err := c.Reader.Read(frame)
        if err != nil {
            log.Println(err)
            log.Println("Stopping reader")
            return
        }

        if n > 0 {
            frame = frame[0:n]
            c.FrameSize = n
            for _, listener := range c.Forwarder {
                listener <- frame
            }
        }
        c.FPS++
    }
}

func (c *Camera) Statistics() {
    ticker := time.NewTicker(time.Second)

    for range ticker.C {
        select {
        case <-c.Done:
            log.Println("Stopping statistics")
            return
        default:
            fmt.Printf("\rfps: %v, size: %v ", c.FPS, c.FrameSize)
            c.FPS = 0
        }
    }
}

server.go

package server

import (
    "apps/picam/hardware"
    "fmt"
    "log"
    "net"
    "strconv"
    "time"
)

type session struct {
    Conn         *net.UDPConn
    Message      chan string
    Password     string
    Address      chan *net.UDPAddr
    ClientAdress *net.UDPAddr
}

type Server struct {
    SecurityRobotState chan string
    ServerState        chan string
}

type monitor struct {
    Conn *net.UDPConn
    Kill chan bool
}

func newMonitor(port string) *monitor {
    return &monitor{
        Conn: bindAddress(port),
        Kill: make(chan bool),
    }
}

func NewServer(robotState chan string, serverState chan string) *Server {
    return &Server{
        SecurityRobotState: robotState,
        ServerState:        serverState,
    }
}

func (m *monitor) detectTimeOut(delay time.Duration, done <-chan bool) {
    buffer := make([]byte, 10)
    m.Conn.SetReadDeadline(time.Now().Add(delay))

    for {
        select {
        case <-done:
            log.Println("Shutting down monitor")
            return
        default:
            n, err := m.Conn.Read(buffer)
            if err != nil {
                log.Println(err)
            }

            if n > 0 {
                m.Conn.SetReadDeadline(time.Now().Add(delay))
            }

            if nerr, ok := err.(net.Error); ok && nerr.Timeout() {
                log.Println("No response")
                m.Kill <- true
                return
            }
        }
    }
}

func (s *Server) Run(port, password *string, frame <-chan []byte, lightLvL <-chan int, done <-chan bool) {
    var robotState string
    light := hardware.High

    session := newSession(port, password)
    go session.listenForClients(done)
    log.Println("Listening for clients on: ", session.Conn.LocalAddr())
    defer session.Conn.Close()

    heartBeatPort := stringPortToInt(*port)
    monitor := newMonitor(fmt.Sprintf(":%v", heartBeatPort+1))
    defer monitor.Conn.Close()

    for {
        select {
        case <-done:
            log.Println("Stopping server")
            return
        case address := <-session.Address:
            if robotState != "recording" && light == hardware.Low {
                hardware.BaseHigh()
            }

            log.Println("New client authenticated from address: ", address)
            go monitor.detectTimeOut(time.Second*5, done)
            session.ClientAdress = address
            s.ServerState <- "streaming"
        case f := <-frame:
            switch session.ClientAdress {
            case nil:
                s.ServerState <- "idle"
                continue
            default:
                _, err := session.Conn.WriteToUDP(f, session.ClientAdress)
                if err != nil {
                    log.Println(err)
                }
            }
        case <-monitor.Kill:
            log.Println("Client timed out")
            if robotState != "recording" {
                hardware.BaseLow()
            }
            session.ClientAdress = nil
        case robotState = <-s.SecurityRobotState:
            continue
        case light = <-lightLvL:
            continue
        //using a default case to force the loop to keep running. Is there a better way?
        default:
            continue
        }
    }
}

func newSession(port, password *string) *session {
    return &session{
        Conn:     bindAddress(*port),
        Message:  make(chan string),
        Password: *password,
        Address:  make(chan *net.UDPAddr),
    }
}

func (s *session) listenForClients(done <-chan bool) {
    for {
        select {
        case <-done:
            log.Println("Stopping listener")
            return
        default:
            buf := make([]byte, 1024)
            n, addr, err := s.Conn.ReadFromUDP(buf)
            if err != nil {
                log.Println("Read error. ", err)
            }

            m := buf[0:n]

            if s.Password == "" {
                s.Address <- addr
                continue
            }

            if s.authenticate(string(m), addr) {
                s.Address <- addr
            }
        }
    }
}

drivers.go

package hardware

import (
    "bufio"
    "fmt"
    "log"
    "os"
    "strings"
    "time"

    "apps/picam/storage"

    rpio "github.com/stianeikeland/go-rpio"
    "golang.org/x/exp/io/spi"
)

const (
    Low = iota
    High
    moving
    still
)

type SecurityRobot struct {
    VideoFeed   chan []byte
    Done        chan bool
    recording   bool
    RobotState  chan string
    ServerState chan string
}

func (r *SecurityRobot) Video(lightLevel <-chan int, motion <-chan int) {
    var serverState string
    var light int
    var m int

    frame := make(chan []byte, 1)
    stop := make(chan bool, 1)

    for {
        select {
        case <-r.Done:
            log.Println("Local recording exit - ")
            return
        case light = <-lightLevel:
            continue
        case m = <-motion:
            continue
        case f := <-r.VideoFeed:
            switch m {
            case still:
                if r.recording {
                    stop <- true
                    r.recording = false
                    r.RobotState <- "idle"
                    if serverState != "streaming" {
                        BaseLow() //light switch OFF
                    }
                }
            case moving:
                if !r.recording {

                    if light == Low {
                        BaseHigh() //light switch ON
                    }

                    r.recording = true
                    r.RobotState <- "recording"

                    now := time.Now()

                    fname := fmt.Sprintf("%s.h264", now.Format("02-01-2006T15:04:05"))
                    fname = strings.Replace(fname, ":", "-", 3)

                    go writeToFile(fname, frame, stop)
                }
                frame <- f
            }
        case serverState = <-r.ServerState:
            continue
        }
    }
}

func writeToFile(fname string, frame <-chan []byte, stop chan bool) {
    file, err := os.Create(fname)
    if err != nil {
        log.Fatal(err)
    }

    fw := bufio.NewWriter(file)

    log.Println("Recording...")
    for {
        select {
        case <-stop:
            fw.Flush()
            file.Close()
            //upload to google drive
            log.Println("Recording stopped")
            log.Println("Uploading to google drive")
            go googleDrive(fname)
            return
        case f := <-frame:
            _, err = fw.Write(f)
            if err != nil {
                log.Println(err)
            }
        }
    }
}

func googleDrive(filename string) {
    id := drive.DriveUpload(filename)
    log.Println("upload success")
    drive.ShareFile(id)
    log.Println("sharing video successful")
    if err := os.Remove(filename); err == nil {
        log.Println("local copy removed")
    }
}

//driver for chip mcp3008
//converts analog data to digitial
func ReadLightLevels(done chan bool) chan int {
    lightLvL := make(chan int, 1)
    go func() {
        var l int //setting default light lvl to dark

        dev, err := spi.Open(&spi.Devfs{
            Dev: "/dev/spidev0.0", Mode: spi.Mode0, MaxSpeed: 1000000})

        if err != nil {
            log.Println("SPI error", err)
            log.Println("Did you forget to enable SPI interface?")
            return
        }

        defer dev.Close()

        tick := time.Tick(time.Second * 1)
        write := []byte{1, 8 << 4, 0}

        for {
            select {
            case <-done:
                log.Println("Light sensor goroutine exit - √")
                return
            case <-tick:
                read := make([]byte, 3)

                if err := dev.Tx(write, read); err != nil {
                    fmt.Println("SPI read/write error. ", err)
                    return
                }

                code := int(read[1]&3)<<8 + int(read[2])

                voltage := (float32(code) * 3.3) / 1024

                if voltage <= 0.750 && l == High {
                    l = Low
                    lightLvL <- Low
                }

                if voltage >= 0.750 && l == Low {
                    l = High
                    lightLvL <- High
                }
            }
        }
    }()

    return lightLvL
}

func DetectMovement(done chan bool) chan int {
    motion := make(chan int, 1)
    tick := time.Tick(time.Microsecond * 10)

    go func() {
        for {
            select {
            case <-done:
                log.Println("PIR detector goroutine exit - √")
                return
            case <-tick:
                switch rpio.ReadPin(pirInput) {
                case rpio.High:
                    motion <- moving
                case rpio.Low:
                    motion <- still
                }
            }
        }
    }()

    return motion
}

func BaseHigh() {
    transistorBase.High()
}

func BaseLow() {
    transistorBase.Low()
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.