diff options
| -rw-r--r-- | cmdiface.erl | 40 | ||||
| -rw-r--r-- | devicemon.erl | 112 | ||||
| -rw-r--r-- | devicemon.hrl | 2 | ||||
| -rw-r--r-- | eventserv.erl | 98 | ||||
| -rw-r--r-- | eventserv.hrl | 1 | ||||
| -rw-r--r-- | fakedev.erl | 27 | ||||
| -rw-r--r-- | portman.erl | 33 | ||||
| -rw-r--r-- | rsbusserv.erl | 119 | ||||
| -rw-r--r-- | startscript.erl | 73 | ||||
| -rw-r--r-- | startserv.erl | 43 | ||||
| -rw-r--r-- | tcpserv.erl | 102 |
11 files changed, 650 insertions, 0 deletions
diff --git a/cmdiface.erl b/cmdiface.erl new file mode 100644 index 0000000..0b7e84c --- /dev/null +++ b/cmdiface.erl @@ -0,0 +1,40 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% +%%% @end +%%% Created : 10. May 2014 下午9:20 +%%%------------------------------------------------------------------- +-module(cmdiface). +-author("Joe Zhao"). + +-compile([debug_info,export_all]). + +%% API +%%-export([]). + +reg_feedbackOps(Addr,Name,T,MsgReq,MsgRes,To) -> + io:format("Logging new command: ~p ~n",[Name]), + Init=fun () -> rsbusserv ! {devicemon,Addr,MsgReq} end, + Exit= + fun () -> + rsbusserv ! {devicemon,Addr,[]} + end, + Pid = spawn(?MODULE,res_manip,[Init,Exit,To]), + Call = + fun () -> + Pid ! rsbusserv:rsbusreq(Addr,MsgRes) + end, + eventserv:reg_future(Call,T). + +invoke_cmd(Addr,Name,MsgReq) -> + io:format("Command: ~p ~n",[Name]), + rsbusserv! {devicemon,comm,Addr,MsgReq}. + +res_manip(Init,Exit,To) -> + Init(), + receive + Res -> To ! Res + end, + Exit().
\ No newline at end of file diff --git a/devicemon.erl b/devicemon.erl new file mode 100644 index 0000000..84c2849 --- /dev/null +++ b/devicemon.erl @@ -0,0 +1,112 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% This modules monitors different devices which includes registering devices when linked and monitoring their states. +%%% Registering devices and deregister them are done automatically when device are on/off line. +%%% Other than that the changing states of different devices is recorded, and boardcast / return when queried. +%%% @end +%%% Created : 29. 四月 2014 15:35 +%%%------------------------------------------------------------------- +-module(devicemon). +-author("Joe Zhao"). + +-compile([debug_info,export_all]). + +-include("devicemon.hrl"). + +%% API +%% -export([]). + +-define(QTOUT,100). + +cleanloop() -> +%% Cleanup the message queue -> Don't call it + receive + _ -> cleanloop() + after 0 -> + ok + end. + +deviceloop(Devices) -> + receive + %% Basic Ops within server + {Pid,shutdown} -> ok; + {Pid,reg,Addr,Type} -> + %% Register -> auto replace + deviceloop(orddict:store(Addr,#device{type=Type},Devices)); + {Pid,reg,Addr,Type,State} -> + deviceloop(orddict:store(Addr,#device{type=Type,state = State},Devices)); + {Pid,dereg,Addr} -> + %% Register -> auto remove + case orddict:find(Addr,Devices) of + error -> deviceloop(Devices); + _Else -> deviceloop(orddict:erase(Addr,Devices)) + end; + {Pid,stat,Addr} -> + case orddict:find(Addr,Devices) of + error -> Pid ! {self(),Addr,error}; + {ok,Value} -> Pid ! {self(),Addr,Value} + end, + deviceloop(Devices); + + %% Query about device list & status + %% Next step: remembering states + {Pid,comm,[0]} -> % General query + Pid ! {self(),comm,orddict:fold(fun(Key,Val,Acc) -> [Key,Val#device.type]++Val#device.state++Acc end, [],Devices)}, + deviceloop(Devices); + {Pid,comm,[0,Addr]} -> % Specific query, this also shouldn't happen. + Dev=orddict:fetch(Addr,Devices), + Pid ! {self(),comm,[Dev#device.type]++Dev#device.state}, + deviceloop(Devices); + + %% Messages received from rsbus + {Pid,device,Addr,{error,Cause}} -> + io:format("[Device Err]Device address: ~p Error: ~p \n",[Addr,Cause]), + self() ! {self(),dereg,Addr}, + deviceloop(Devices); + {Pid,device,Addr,{stat,Type,Msg}} -> + tcpserv ! {self(),comm,[Addr,Type|Msg]}, + deviceloop(orddict:store(Addr,#device{type=Type,state = Msg},Devices)); + + %% Messages received from some other server -> redirect to rsbus + {Pid,comm,[Addr|Msg]} -> + rsbusserv ! {self(),comm,Addr,Msg}, + deviceloop(Devices); + + %% General ops + {Pid,clean} -> % Clean message queue -> use it when garbage might be cloting the server + cleanloop(), + deviceloop(Devices); + {Pid,heartbeat,Addr} -> % Specific heartbeat !!! Good stuff + self() ! {self(),comm,[Addr]}, + deviceloop(Devices); + {Pid,heartbeat} -> % This type of heartbeat will cause congestion, use with caution + [self() ! {self(),comm,[Addr]}|| Addr <- lists:seq(1,?ADDRANGE)], + deviceloop(Devices) + end. + +devicemon() -> + register(?MODULE,self()), + spawn(?MODULE,devicepool,[?ADDRANGE]), + deviceloop(orddict:new()). + +getStat(Addr) -> + devicemon ! {self(),stat,Addr}, + receive + {_Pid,Addr,error} -> error; + {_Pid,Addr,Val} -> Val + after ?QTOUT -> + error + end. + +getStaus(Addr) -> + case getStat(Addr) of + error -> error; + Val -> Val + %% device status multiplexer + %% ready | busy + end. + +start() -> + spawn(?MODULE,devicemon,[]).
\ No newline at end of file diff --git a/devicemon.hrl b/devicemon.hrl new file mode 100644 index 0000000..b983b5d --- /dev/null +++ b/devicemon.hrl @@ -0,0 +1,2 @@ +-record(device,{type,state=[0,0,0,0]}). +-define(ADDRANGE,10).
\ No newline at end of file diff --git a/eventserv.erl b/eventserv.erl new file mode 100644 index 0000000..dc4dcde --- /dev/null +++ b/eventserv.erl @@ -0,0 +1,98 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% This module serves the purpose of registering and monitoring different events +%%% Whether they are periodic or oneshot. +%%% Events are happening in a sense that none should be depending on those that happend before. +%%% Spawning one with mutable state is thus not recommended and not supported in such modules. +%%% Thus, the events are defined so that none of them would have states change during different invocations. +%%% @end +%%% Created : 27. 四月 2014 19:11 +%%%------------------------------------------------------------------- +-module(eventserv). +-author("Joe Zhao"). + +-include("eventserv.hrl"). +-compile([debug_info,export_all]). + +%% API +%%-export([]). + +event(Pid,S) -> + receive + {_PPid,shutdown} -> + io:format("I, ~p, am killed.\n",[S#event.name]) + after S#event.period -> + Pid ! {self(),run,S}, + if + S#event.spawn -> spawn(S#event.call,[self()]); + true -> (S#event.call)(self()) + end, + case S#event.type of + oneshot -> ok; + periodic -> event(Pid,S); + _Else -> ok + end + end. + +recloop(Events) -> + receive + {Pid,shutdown} -> [ Proc ! {self(),shutdown} || Proc<- orddict:fetch_keys(Events)]; + {PPid,reg,S} -> + Pid = spawn(?MODULE,event,[self(),S]), + link(Pid), + recloop(orddict:store(Pid,S,Events)); + {Pid,run,S} -> + io:format("Event run, type ~p: ~p\n",[S#event.type,S#event.name]), + recloop(Events); + {'EXIT',Pid,Cond} when Cond==normal;Cond==shutdown -> + io:format("Event died naturally, pid: ~p\n.",[Pid]), + recloop(orddict:erase(Pid,Events)); + {'EXIT',Pid,Cause} -> + S=orddict:fetch(Pid,Events), + io:format("Event died abruptly(~p), pid: ~p ; name: ~p\n.",[Cause,Pid,S#event.name]), + if + S#event.type == periodic -> + io:format("Event(periodic) will restart, name: ~p\n.",[S#event.name]), + self() ! {self(),reg,S} + end, + recloop(orddict:erase(Pid,Events)) + end. + +evemon(Events) -> + process_flag(trap_exit,true), + register(?MODULE,self()), + recloop(Events). + +start() -> + spawn(?MODULE,evemon,[orddict:new()]). + +reg_future(Name,Call,T) -> + eventserv ! {self(),reg,#event{name=Name,server=eventserv, + call=Call,period=T,type=oneshot}}. + +reg_future(Call,T) -> + eventserv ! {self(),reg,#event{name="Unamed",server=eventserv, + call=Call,period=T,type=oneshot}}. + +reg_periodic(Name,Call,T) -> + eventserv ! {self(),reg,#event{name=Name,server=eventserv, + call=Call,period=T}}. + +reg_periodic(Call,T) -> + eventserv ! {self(),reg,#event{name="Unamed",server=eventserv, + call=Call,period=T}}. + +regevent(Pid,S) -> + Pid ! {self(),reg,S}. + +test() -> + Pid=start(), + Pid ! {self(),reg,#event{name="test1",server=Pid, + call=fun (PPid) -> io:format("HAHA im here!!!!\n") end,period=1000}}, + timer:sleep(500), + Pid ! {self(),reg,#event{name="test2",server=Pid, + call=fun (PPid) -> io:format("HAHA im away and buggy!!!!\n"),exit("I just love to die") end,period=1000}}, + timer:sleep(10000), + Pid ! {self(),shutdown}.
\ No newline at end of file diff --git a/eventserv.hrl b/eventserv.hrl new file mode 100644 index 0000000..4a53e15 --- /dev/null +++ b/eventserv.hrl @@ -0,0 +1 @@ +-record(event,{name="",server,call,type=periodic,period,spawn=true}).
\ No newline at end of file diff --git a/fakedev.erl b/fakedev.erl new file mode 100644 index 0000000..20f1311 --- /dev/null +++ b/fakedev.erl @@ -0,0 +1,27 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% +%%% @end +%%% Created : 29. 四月 2014 19:14 +%%%------------------------------------------------------------------- +-module(fakedev). +-author("Joe Zhao"). + +-compile([export_all,debug_info]). + +%% API +%% -export([]). + +devmon() -> + ok. + +stopdev(Addr) -> + ok. + +startdev(Addr) -> + ok. + +start() -> + spawn(?MODULE,devmon(),[orddict:new()]).
\ No newline at end of file diff --git a/portman.erl b/portman.erl new file mode 100644 index 0000000..393a566 --- /dev/null +++ b/portman.erl @@ -0,0 +1,33 @@ +%%%------------------------------------------------------------------- +%%% @author Tuowen +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% +%%% @end +%%% Created : 26. 四月 2014 16:42 +%%%------------------------------------------------------------------- +-module(portman). +-author("Tuowen"). + +%% API +-export([start/1,serv/1,send/2]). + +oploop(Sock) -> + receive + {Pid,shutdown} -> ok; + {Pid,Msg} -> + gen_tcp:send(Sock,Msg), + oploop(Sock) + end. + +serv({Host,Port}) -> + {ok, Sock} = gen_tcp:connect(Host, Port, + [list, {packet, 1}]), + oploop(Sock), + ok = gen_tcp:close(Sock). + +start(Config) -> + spawn(?MODULE,serv,[Config]). + +send(Pid,Msg) -> + Pid ! {self(),Msg}.
\ No newline at end of file diff --git a/rsbusserv.erl b/rsbusserv.erl new file mode 100644 index 0000000..997e590 --- /dev/null +++ b/rsbusserv.erl @@ -0,0 +1,119 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% This server monitors behaviors on the serial bus. +%%% Whose duty includes serialize queries, message encoding, +%%% and response handling(only when slave request repitition of the previous command) +%%% The returns of the command are sent back to its origins(first parameter of the tuple, can also be someone else). +%%% @end +%%% Created : 27. 四月 2014 16:52 +%%%------------------------------------------------------------------- +-module(rsbusserv). +-author("Joe Zhao"). + +-compile([debug_info,export_all]). + +%% API +%% -export([start/0]). + +-define(DEVICE,"/dev/ttyUSB0"). +-define(SPEED,19200). +-define(TIMEOUT,40). % Timeout in milisec +-define(REQTIMEOUT,200). + +-define(STARTBIT,[16#5B,16#AD]). +-define(STOPBIT,[16#A4,16#52]). +-define(REP,16#FF). + +rsbuslistener(Pid) -> + SerialPort = serial:start([{speed,?SPEED},{open,?DEVICE}]), + register(?MODULE,self()), + rsbus_listner(Pid,SerialPort). + +rsbus_parity(Msg) -> + lists:foldl(fun (X,P) -> X bxor P end,0,Msg). + +rsbus_rec([],ok,Msg,[]) -> + {stat,lists:reverse(tl(Msg)),hd(Msg)}; +rsbus_rec([H1,H2|Tail],read,Msg,[H1,H2]) -> + rsbus_rec([],ok,Msg,[]); +rsbus_rec([H1,H2,H3|Tail],read,Msg,Cmp) -> + rsbus_rec([H2,H3|Tail],read,[H1|Msg],Cmp); +rsbus_rec([H1,H2,H3,H4|Tail1],wait,[],[H1,H2,H3,H4]) -> + rsbus_rec(Tail1,read,[],?STOPBIT); +rsbus_rec([H|Tail],wait,[],Cmp) -> + rsbus_rec(Tail,wait,[],Cmp); +rsbus_rec(Rec,State,Msg,Cmp) -> + receive + {data, Bytes} -> + erlang:display(binary_to_list(Bytes)), + rsbus_rec(Rec++binary_to_list(Bytes),State,Msg,Cmp) + after ?TIMEOUT -> + {error,timeout} + end. + +rsbus_send(SerialPort,Addr,Msg) -> + %% Return {stat,Type,Stat} upon success + %% else return {error,timeout|parse} + SerialPort ! {send,?STARTBIT}, + SerialPort ! {send,[Addr]}, + SerialPort ! {send,Msg}, + P=rsbus_parity(Msg) bxor Addr, + SerialPort ! {send,[P]}, + SerialPort ! {send,?STOPBIT}, + io:format("Send Complete!\n"), + case rsbus_rec([],wait,[],?STARTBIT++[0,Addr]) of + {stat,[?REP],RP} -> rsbus_send(SerialPort,Addr,Msg); % Will resend no matter what + {stat,Stat,RP} -> + erlang:display(RP), + TestP =rsbus_parity([RP|Stat]), + if + TestP /= Addr -> {err,parse}; + true -> {stat,hd(Stat),tl(Stat)} + end; + {error,timeout} -> {error,timeout} + end. + + +%% This function can effectively serialize access +rsbus_listner(Pid,SerialPort) -> + erlang:display(self()), + receive + {data, _Bytes} -> + %% This shouldn't happen, no process should initiate comm + %% Discard everything + rsbus_listner(Pid,SerialPort); + {PPid,comm,Addr,Msg} -> + io:format("RSBus sending..."), + PPid ! {self(),device,Addr,rsbus_send(SerialPort,Addr,Msg)}, % Process and return + rsbus_listner(Pid,SerialPort); + {Pid,shutdown} -> + SerialPort ! stop, + ok + end. + +rsbusloop(Pid) -> + receive + {'EXIT',Pid,Cond} when Cond == normal; Cond==shutdown -> rsbusloop(spawn_link(?MODULE,rsbuslistener,[self()])); + {PPid,shutdown} -> Pid ! {self(),shutdown} + end. + +rsbusmon() -> + process_flag(trap_exit,true), + Pid = spawn_link(?MODULE,rsbuslistener,[self()]), + rsbusloop(Pid). + +rsbusreq(Addr,Msg) -> + rsbusserv ! {self(),comm,Addr,Msg}, + receive + {PPid,device,Addr,{error,Reason}} -> + {error,Reason}; + {PPid,device,Addr,{stat,Type,Msg}} -> + {stat,Type,Msg} + after ?REQTIMEOUT -> + {error,timeout} + end. + +start() -> + spawn(?MODULE,rsbusmon,[]).
\ No newline at end of file diff --git a/startscript.erl b/startscript.erl new file mode 100644 index 0000000..a281211 --- /dev/null +++ b/startscript.erl @@ -0,0 +1,73 @@ +%%%------------------------------------------------------------------- +%%% @author joe +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% +%%% @end +%%% Created : 15. May 2014 下午2:29 +%%%------------------------------------------------------------------- +-module(startscript). +-author("joe"). + +-compile([debug_info,export_all]). +%% API +%% -export([]). + +-include("devicemon.hrl"). + +-define(HBPERIOD,500). +-define(SCANPERIOD,2000). + +busbeater(Pid,Addr) -> + case devicemon:getStaus(Addr) of + error -> + spawnscanner(Addr), + Pid ! {self(),shutdown}; + busy -> ok; + ready -> + case rsbusserv:rsbusreq(Addr,[]) of + {stat,Type,Stat} -> + devicemon ! {self(),reg,Addr,Type,Stat}; + {error,Cause} -> + devicemon ! {self(),dereg,Addr}, + spawnscanner(Addr), + Pid ! {self(),shutdown} + end + end. + +busscanner(Pid,Addr) -> + %% if devicemon:getStat(Addr) + %% Add device status multiplexer here + case devicemon:getStaus(Addr) of + error -> + case rsbusserv:rsbusreq(Addr,[]) of + {stat,Type,Stat} -> + devicemon ! {self(),reg,Addr,Type,Stat}, + spawnbeater(Addr), + Pid ! {self(),shutdown}; + {error,Cause} -> + devicemon ! {self(),dereg,Addr} + end; + _Else -> + spawnbeater(Addr), + Pid ! {self(),shutdown} + end. + +spawnscanner(Addr) -> + eventserv:reg_periodic("Scanner",fun (Pid) -> busscanner(Pid,Addr) end,?SCANPERIOD). +spawnbeater(Addr) -> + eventserv:reg_periodic("Heartbeat",fun (Pid) -> busbeater(Pid,Addr) end,?HBPERIOD). + +runrange(Ed,Ed,Fun) -> + Fun(Ed); +runrange(St,Ed,Fun) -> + Fun(St), + runrange(St+1,Ed,Fun). + +start() -> + Tcp=tcpserv:start(5575), + Rsbus=rsbusserv:start(), + Evserv=eventserv:start(), + Devmon=devicemon:start(), + runrange(1,10,fun (Addr)->spawnscanner(Addr) end). + diff --git a/startserv.erl b/startserv.erl new file mode 100644 index 0000000..2f583a4 --- /dev/null +++ b/startserv.erl @@ -0,0 +1,43 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% +%%% @end +%%% Created : 26. 四月 2014 16:34 +%%%------------------------------------------------------------------- +-module(startserv). +-author("Joe Zhao"). + +-compile([debug_info]). + +%% API +-export([start/0,startmon/0,mon/0]). + +mon() -> + process_flag(trap_exit,true), + Pid = spawn_link(tcpport,start,[5575]), + receive + {_,shutdown} -> Pid ! shutdown; + {'EXIT',Pid,normal} -> ok; + {'EXIT',Pid,shutdown} -> ok; + {'EXIT',Pid,_} -> mon() + end. + +startmon() -> + spawn(?MODULE,mon,[]). + +start() -> + PPid=startmon(), + timer:sleep(100), + Pid=portman:start({"localhost",5575}), + portman:send(Pid,["Something 1"]), + timer:sleep(100), + portman:send(Pid,["Something 2"]), + timer:sleep(100), + portman:send(Pid,["Something 3"]), + timer:sleep(100), + portman:send(Pid,["Something 4"]), + timer:sleep(100), + portman:send(Pid,shutdown), + PPid ! {self(),shutdown}. diff --git a/tcpserv.erl b/tcpserv.erl new file mode 100644 index 0000000..91a25a6 --- /dev/null +++ b/tcpserv.erl @@ -0,0 +1,102 @@ +%%%------------------------------------------------------------------- +%%% @author Joe Zhao +%%% @copyright (C) 2014, <COMPANY> +%%% @doc +%%% TCP server for socket connections +%%% One listenners of ports +%%% Spawns one receiver and one sender for different socket. +%%% @end +%%% Created : 27. 四月 2014 20:52 +%%%------------------------------------------------------------------- +-module(tcpserv). +-author("Joe Zhao"). + +-compile([debug_info,export_all]). + +-define(TCP_OPTIONS,[list, {packet, 1}, {active, false}]). +-define(DLYTIME,1000). +%% API +%%-export([]). + +lsockinit() -> ok. + +do_send(Sock,Pid,From) -> + receive + {From,exiting} -> ok; + {Pid,shutdown} -> + gen_tcp:close(Sock), + ok; + {_PPid,comm,Msg} -> + gen_tcp:send(Sock,Msg), + do_send(Sock,Pid,From) + end. + +do_recv(Sock,Pid,To,Cnt) -> + case gen_tcp:recv(Sock,0) of + {ok, B} -> + io:format("~p:\n ~p\n",[Cnt,B]), + manip_msg(Pid,To,B), + %% devicemon ! {self(),comm,Msg}, + do_recv(Sock,Pid,To,Cnt+1); + {error, closed} -> + ok + end. + +%% Message manipulation & dispatch routine +manip_msg(Pid,To,[0,0,Addr,MsgReq,MsgRes|Name]) -> + %% Operation command dispatch -> Complex(devicemon + Message handler) + cmdiface:reg_feedbackOps(Addr,Name,?DLYTIME,MsgReq,MsgRes,To); +manip_msg(Pid,To,[0|Tail]) -> + %% Seperation from the next part + devicemon ! [0|Tail]; +manip_msg(Pid,To,[Addr,Cmd|Name]) when Addr<10 -> + %% Seperation from the next part + cmdiface:invoke_cmd(Addr,Name,[Cmd]); +manip_msg(Pid,To,Msg) -> % This needs improvements + %% Fallback -> WTF + Pid ! {self(),comm,Msg}. + +createsock(Sock,Pid) -> + To=spawn(?MODULE,do_send,[Sock,Pid,self()]), + Pid ! {self(),create,To}, + ok = do_recv(Sock,Pid,To,0), + To ! Pid ! {self(),exiting}. + +listenport(LSock,Pid) -> + {ok,Sock} = gen_tcp:accept(LSock), + spawn(?MODULE,createsock,[Sock,Pid]), + listenport(LSock,Pid). + +portlistener(Port,Pid) -> + {ok, LSock} = gen_tcp:listen(Port, [list, {packet, 1}, {active, false}]), + listenport(LSock,Pid). + + +monloop(Procs) -> + receive + {_Pid,shutdown} -> + [ To ! shutdown || {_From,To} <- orddict:to_list(Procs) ]; + {Pid,create,To} -> + monloop(orddict:store(Pid,To,Procs)); + {Pid,exiting} -> + monloop(orddict:erase(Pid,Procs)); + {_Pid,comm,Msg} -> + [ To ! {self(),comm,Msg} || {_From,To} <- orddict:to_list(Procs) ], + monloop(Procs) + end. + +tcpmon(Port) -> + Lis=spawn(?MODULE,portlistener,[Port,self()]), + register(?MODULE,self()), + monloop(orddict:new()), + exit(Lis,shutdown). + +start(Port) -> + spawn(?MODULE,tcpmon,[Port]). + + +trap() -> trap(). + +test() -> + start(5575), + trap().
\ No newline at end of file |