FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
cdds-python/f05log.py at master · atolab/cdds-python · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
atolab
/
cdds-python
Public
Notifications
You must be signed in to change notification settings
Fork
4
Star
11
Code
Issues
1
Pull requests
1
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
cdds-python
/
f05log.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
103 lines (78 loc) · 2.94 KB
Breadcrumbs
cdds-python
/
f05log.py
Copy path
File metadata and controls
103 lines (78 loc) · 2.94 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
from
cdds
import
*
from
fog05
.
interfaces
.
Types
import
*
import
sys
def
cache_discovered
(
dr
):
for
(
s
,
i
)
in
dr
.
take
(
all_samples
()):
if
i
.
valid_data
:
print
(
str
(
s
))
def
cache_discovery_event
(
r
,
s
):
print
(
"Cache discovery event"
)
def
log_samples
(
dr
):
for
(
s
,
i
)
in
dr
.
take
(
all_samples
()):
if
i
.
valid_data
:
print
(
str
(
s
))
def
start_tlog
(
root
):
dds_runtime
=
Runtime
.
get_runtime
()
dp
=
Participant
(
0
)
pub
=
Publisher
(
dp
,
root
)
sub
=
Subscriber
(
dp
,
root
)
store_info_topic
=
FlexyTopic
(
dp
,
"FOSStoreInfo"
)
key_value_topic
=
FlexyTopic
(
dp
,
"FOSKeyValue"
)
store_info_writer
=
FlexyWriter
(
pub
,
store_info_topic
,
DDS_State
)
store_info_reader
=
FlexyReader
(
sub
,
store_info_topic
,
log_samples
,
DDS_State
)
store_info_reader
.
on_liveliness_changed
(
cache_discovery_event
)
key_value_writer
=
FlexyWriter
(
pub
,
key_value_topic
,
DDS_State
)
key_value_reader
=
FlexyReader
(
sub
,
key_value_topic
,
log_samples
,
DDS_State
)
miss_topic
=
FlexyTopic
(
dp
,
"FOSStoreMiss"
)
miss_writer
=
FlexyWriter
(
pub
,
miss_topic
,
DDS_Event
)
miss_reader
=
FlexyReader
(
sub
,
miss_topic
,
log_samples
,
DDS_Event
)
hit_topic
=
FlexyTopic
(
dp
,
"FOSStoreHit"
)
hit_writer
=
FlexyWriter
(
pub
,
hit_topic
,
DDS_Event
)
hit_reader
=
FlexyReader
(
sub
,
hit_topic
,
log_samples
,
DDS_Event
)
missmv_topic
=
FlexyTopic
(
dp
,
"FOSStoreMissMV"
)
missmv_writer
=
FlexyWriter
(
pub
,
missmv_topic
,
DDS_Event
)
missmv_reader
=
FlexyReader
(
sub
,
missmv_topic
,
log_samples
,
DDS_Event
)
hitmv_topic
=
FlexyTopic
(
dp
,
"FOSStoreHitMV"
)
hitmv_writer
=
FlexyWriter
(
pub
,
hitmv_topic
,
DDS_Event
)
hitmv_reader
=
FlexyReader
(
sub
,
hitmv_topic
,
log_samples
,
DDS_Event
)
if
__name__
==
'__main__'
:
if
len
(
sys
.
argv
)
>
1
:
start_tlog
(
sys
.
argv
[
1
])
print
(
"Press 'x' or 'X' to exit..."
)
done
=
False
while
done
is
False
:
c
=
input
()
if
c
.
capitalize
()
==
"X"
:
done
=
True
else
:
print
(
"USAGE:
\n
\t
flog"
)
Back
|
FazBrowse Home
|
New Git URL