-
Notifications
You must be signed in to change notification settings - Fork 0
/
all_functions.py
2468 lines (2201 loc) · 108 KB
/
all_functions.py
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
__author__ = "David Katz-Wigmore"
__version__ = ".1"
import pandas as pd
import numpy as np
from datetime import datetime
from dateutil.relativedelta import relativedelta
class MetricsFunctions:
def average_los_in_es_shelter(self, entries_df, cleaned=False):
"""
Used For:
:param entries_df:
:return:
"""
stays = entries_df[
(
entries_df["Entry Exit Provider Id"].str.contains("Hansen") |
entries_df["Entry Exit Provider Id"].str.contains("Columbia") |
entries_df["Entry Exit Provider Id"].str.contains("Willamette") |
entries_df["Entry Exit Provider Id"].str.contains("SOS") |
entries_df["Entry Exit Provider Id"].str.contains("5th")
)
]
stays["Entry Date"] = pd.to_datetime(stays["Entry Exit Entry Date"]).dt.date
stays["Exit Date"] = pd.to_datetime(stays["Entry Exit Exit Date"]).dt.date
stays["Exit Date"].fillna(value=datetime(year=2017, month=9, day=30, hour=23, minute=59, second=59))
stays["LOS"] = (stays["Exit Date"] - stays["Entry Date"]).dt.days
filtered_stays = stays[["Client Uid", "Entry Date", "Exit Date", "LOS"]]
total_days = filtered_stays["LOS"].sum()
total_stays = len(filtered_stays.index)
un_cleaned_mean = filtered_stays["LOS"].mean()
cleaned_los_data = filtered_stays[
np.abs(filtered_stays.LOS - filtered_stays.LOS.mean()) <= (3 * filtered_stays.LOS.std(ddof=0))
]
cleaned_mean = cleaned_los_data.mean()
if cleaned:
return tuple(["Length of stay per emergency shelter resident (days)", "", cleaned_mean])
else:
return tuple(["Length of stay per emergency shelter resident (days)", "", un_cleaned_mean])
def average_los_in_res_shelter(self, entries_df, cleaned=False):
"""
Used For: Agency
:param entries_df: data frame of all entries report
:param cleaned: if cleaned == True method will return the mean excluding outliers, which are defined as 3
standard deviations from the norm
:return:
"""
stays = entries_df[
(
entries_df["Entry Exit Provider Id"].str.contains("Doreen's") |
entries_df["Entry Exit Provider Id"].str.contains("Clark Center") |
entries_df["Entry Exit Provider Id"].str.contains("Jean's")
)
]
stays["Entry Date"] = pd.to_datetime(stays["Entry Exit Entry Date"]).dt.date
stays["Exit Date"] = pd.to_datetime(stays["Entry Exit Exit Date"]).dt.date
stays["Exit Date"].fillna(value=datetime(year=2017, month=9, day=30, hour=23, minute=59, second=59))
stays["LOS"] = (stays["Exit Date"] - stays["Entry Date"]).dt.days
filtered_stays = stays[["Client Uid", "Entry Date", "Exit Date", "LOS"]]
total_days = filtered_stays["LOS"].sum()
total_stays = len(filtered_stays.index)
un_cleaned_mean = filtered_stays["LOS"].mean()
cleaned_los_data = filtered_stays[
np.abs(filtered_stays.LOS - filtered_stays.LOS.mean()) <= (3 * filtered_stays.LOS.std())
]
cleaned_mean = cleaned_los_data.mean()
if cleaned:
return tuple(["Length of stay per residential shelter resident (days)", "", cleaned_mean])
else:
return tuple(["Length of stay per residential shelter resident (days)", "", un_cleaned_mean])
def calculate_average_wait_list_length(self, waitlist_df, waitlist="Men"):
"""
Used by: Agency
Used for: LOS on Men's Waitlist; LOS on Women's Waitlist
:param waitlist: This should be either 'Men' or 'Women'
:param waitlist_df:
:return:If there are no values that result from a removal of outlier items (3 standard deviations from the mean)
then a basic mean will be returned other wise the return value will exclude outliers.
"""
def check_list(waitlist, data_frame=waitlist_df):
if waitlist == "Men":
specific_wait_list = data_frame[data_frame["Waitlist Name"].str.contains(waitlist)]
return specific_wait_list
elif waitlist == "Women":
specific_wait_list = data_frame[
(
data_frame["Waitlist Name"].str.contains(waitlist) |
data_frame["Waitlist Name"].str.contains("Jean's")
)
]
return specific_wait_list
specified = check_list(waitlist, waitlist_df.dropna(axis=0, subset=["Waitlist Name"]))
specified["Event Date"] = pd.to_datetime(specified["Waitlist Event Date"])
filtered = specified[["ClientID", "Event Date", "Waitlist Event Code"]]
filtered["q_start"] = datetime(year=2017, month=10, day=1, hour=0, minute=0, second=0)
filtered["q_end"] = datetime(year=2017, month=12, day=31, hour=23, minute=59, second=59)
in_shelter = filtered[
(
filtered["Waitlist Event Code"].str.contains("IN") &
(filtered["Event Date"] >= filtered["q_start"]) &
(filtered["Event Date"] <= filtered["q_end"])
)
]
in_shelter["In Date"] = in_shelter["Event Date"]
in_shelter_clean = in_shelter[["ClientID", "In Date"]]
new_to_list = filtered[
(
filtered["Waitlist Event Code"].str.contains("NEW") &
filtered["ClientID"].isin(in_shelter["ClientID"])
)
]
new_to_list_sorted = new_to_list.sort_values(by="Event Date").drop_duplicates(subset="ClientID", keep="first")
new_to_list_clean = new_to_list_sorted[["ClientID", "Event Date"]]
merged = in_shelter_clean.merge(new_to_list_clean, on="ClientID", how="left")
merged["time_on_list"] = (merged["In Date"] - merged["Event Date"]).dt.days
all_values_mean = merged.time_on_list.mean()
clean_mean = merged[
np.abs(merged.time_on_list - merged.time_on_list.mean()) <= (3 * merged.time_on_list.std())
].time_on_list.mean()
if clean_mean:
return tuple([
"Length of {}'s shelter waitlist (in days)".format(waitlist),
"",
"{} Days".format(clean_mean)
])
else:
return tuple([
"Length of {}'s shelter waitlist (in days)".format(waitlist),
"",
"{} Days".format(all_values_mean)
])
def count_access_employment_services(self, services_df, staff_on_team, direct=False):
"""
prior to 7/1/2017 = ["Charles Oneill(7792)"]
from 7/1/2017 - 10/1/2017 = ["Karla Smith(8044)", "Charles Oneill(7792)"]
post 10/1/2017 = ["Randell Phillips(7727)", "Karla Smith(8044)", "Charles Oneill(7792)"]
Numbers from this report for the quarters run separately then merged and de-duplicated are not matching numbers
run for multiple quarters. This is strange and seems to indicate an error somewhere. Set unit tests and
investigate.
Currently, due to staffing changes during the previous quarter I am running this for multiple quarters using the
direct=True parameter, assigning the output for each quarter's data to a different variable. I am then merging
the variables on the Client Uid column as an 'outer' merge, de-duplicating, and returning the len of the
resulting data frame's index.
:param services_df:
:param staff_on_team:
:return:
"""
if direct == True and type(staff_on_team) == list:
return services_df[
services_df["Service User Creating"].isin(staff_on_team) &
services_df["Service Provide Provider"].str.contains("Support")
].drop_duplicates(subset="Client Uid")
elif direct == True and type(staff_on_team) == str:
return services_df[
services_df["Service User Creating"].str.contains(staff_on_team) &
services_df["Service Provide Provider"].str.contains("Support")
].drop_duplicates(subset="Client Uid")
elif type(staff_on_team) == list:
served = services_df[
services_df["Service User Creating"].isin(staff_on_team) &
services_df["Service Provide Provider"].str.contains("Support")
].drop_duplicates(subset="Client Uid")
return tuple(["600 participants will access employment services", 600, len(served.index)])
elif type(staff_on_team) == str:
served = services_df[
(services_df["Service User Creating"] == staff_on_team) &
services_df["Service Provide Provider"].str.contains("Support")
].drop_duplicates(subset="Client Uid")
return tuple(["600 participants will access employment services", 600, len(served.index)])
else:
return tuple(["Error", "Error", "Error"])
def count_employment_services(self, employment_df, start_date, end_date, metric):
"""
Currently this method uses some rather flawed data for reporting individuals served. Instead please used the
count_access_employment_services method to report on this metric.
The end date should be the last day of the quarter or reporting period.
:param employment_df: A data-frame made from the employment tracker access datasheet available on tprojects.info
on the support services page.
:param start_date: date string using the standard U.S. mm/dd/YYYY format
:param end_date: date string using the standard U.S. mm/dd/YYYY format
:return: An integer indidcating the number of unique individuals served by the employment a
"""
start = datetime.strptime(start_date, "%m/%d/%Y")
end = datetime.strptime(end_date, "%m/%d/%Y")
employment_df["Start Date"] = start
employment_df["End Date"] = end
if metric.lower() == "served":
in_period_access = employment_df[
(
(employment_df["Start Date"] <= employment_df["Created"]) &
(employment_df["Created"] <= employment_df["End Date"])
) |
(
employment_df["Date Income Changed"].notna() &
(
(employment_df["Start Date"] <= employment_df["Date Income Changed"]) &
(employment_df["Date Income Changed"] <= employment_df["End Date"])
)
)
]
served = len(in_period_access.drop_duplicates(subset="PT ID").index)
return tuple(["600 participants will access employment services", 600, served])
elif metric.lower() == "employment":
in_period_gained = employment_df[
employment_df["Employment Gained"].notna() &
(
(employment_df["Start Date"] <= employment_df["Employment Gained"]) &
(employment_df["Employment Gained"] <= employment_df["End Date"])
)
]
gained = len(in_period_gained.drop_duplicates(subset="PT ID").index)
return tuple(["","", gained])
elif metric.lower() == "income":
in_period_gained = employment_df[
employment_df["Date Income Changed"].notna() &
(
(employment_df["Start Date"] <= employment_df["Employment Gained"]) &
(employment_df["Date Income Changed"] <= employment_df["End Date"])
)
]
gained = len(in_period_gained.drop_duplicates(subset="PT ID").index)
return tuple(["15% of participants will increase their incomes ****", "15%", gained])
else:
return tuple(["Error", "Error", "Error"])
def count_employment_services_by_provider(self, employment_df, entries_df, provider):
entries = entries_df[
entries_df["Entry Exit Provider Id"].str.contains(provider)
].drop_duplicates(subset="Client Uid")
employment = employment_df[["PT ID", "Employment Gained", "Date Income Changed", "Employment Lost"]]
merged = entries.merge(employment, left_on="Client Uid", right_on="PT ID", how="left")
cleaned = merged[
(
merged["Employment Gained"].notna() &
(merged["Entry Exit Entry Date"] <= merged["Employment Gained"]) &
(merged["Employment Gained"] <= merged["Entry Exit Exit Date"])
) |
(
merged["Date Income Changed"].notna() &
(merged["Entry Exit Entry Date"] <= merged["Date Income Changed"]) &
(merged["Date Income Changed"] <= merged["Entry Exit Exit Date"])
)
]
return tuple([
"",
"",
"{} / {} = {}%".format(
len(cleaned.index), len(entries.index), 100 * (len(cleaned.index) / len(entries.index))
)
])
def count_all_ep(self, placements_df):
"""
Used by: Agency
:param placements_df:
:return:
"""
ep_placements = placements_df[
placements_df["Intervention Type (TPI)(8745)"] == "Eviction Prevention"
]
de_duplicated_ep = len(ep_placements.drop_duplicates(subset="Client Uid").index)
return tuple(["124 participants will have their evictions prevented", 124, de_duplicated_ep])
def count_all_pp(self, placements_df):
"""
Used by: Agency
:param placements_df:
:return:
"""
pp_placements = placements_df[
placements_df["Intervention Type (TPI)(8745)"].notnull &
placements_df["Intervention Type (TPI)(8745)"].str.contains("Permanent")
]
de_duplicated_pp = len(pp_placements.drop_duplicates(subset="Client Uid").index)
return tuple(["1,065 participants will be permanently housed*", 1065, de_duplicated_pp])
def count_all_placed(self, placements_df):
"""
:param placements_df:
:return:
"""
return placements_df.drop_duplicates(subset="Client Uid")
def count_all_placed_by_provider(
self,
placements_df,
provider=["ACCESS", "SSVF - TPI", "Retention", "Residential CM"]
):
"""
Used By: ACCESS, SSVF, Retention, Residential CM
:param placements_df:
:param provider:
:return:
"""
de_duplicated = placements_df[
placements_df["Department Placed From(3076)"].isin(provider)
].drop_duplicates(subset="Client Uid")
return len(de_duplicated.index)
def count_entries_by_provider(self, entries_df, provider):
"""
Used By:
:param entries_df:
:param provider:
:return:
"""
if type(provider) == str:
count = len(entries_df[
entries_df["Entry Exit Provider Id"].str.contains(provider)
].drop_duplicates(subset="Client Uid", keep="first").index)
if provider == "Residential":
return tuple(["Engage 900 participants in case management", 900, count])
elif provider == "Retention":
pass
elif provider == "SSVF":
pass
elif provider == "ACCESS":
return tuple(["Engage 1200 participants in case management", 1200, count])
elif provider.lower() == "columbia":
return tuple(["700 unduplicated participants have a safe place to sleep", 700, count])
elif provider.lower() == "wil":
return tuple(["1000 unduplicated participants have a safe place to sleep", 1000, count])
elif provider.lower() == "5th":
return tuple(["700 unduplicated participants have a safe place to sleep", 700, count])
elif provider.lower() == "han":
return tuple(["1000 unduplicated participants have a safe place to sleep", 1000, count])
elif provider.lower() == "sos":
return tuple(["700 unduplicated participants have a safe place to sleep", 700, count])
elif (provider.lower() == "clark center") or (provider.lower() == "doreen's"):
return tuple([
"550 unduplicated participants have a safe place to sleep at {}".format(provider),
550,
count
])
elif provider.lower() == "jean's place":
return tuple([
"350 unduplicated participants have a safe place to sleep at Jean's Place",
350,
count
])
elif type(provider) == list:
full_provider_name = {
"cc": "Transition Projects (TPI) - Clark Center - SP(25)",
"col": "Transition Projects (TPI) - Columbia Shelter(5857)",
"dp": "Transition Projects (TPI) - Doreen's Place - SP(28)",
"h": "Transition Projects (TPI) - Hansen Emergency Shelter - SP(5588)",
"jp": "Transition Projects (TPI) - Jean's Place L1 - SP(29)",
"sos": "Transition Projects (TPI) - SOS Shelter(2712)",
"dpgpd": "Transition Projects (TPI) - VA Grant Per Diem (inc. Doreen's Place GPD) - SP(3189)",
"wc": "Transition Projects (TPI) - Willamette Center(5764)",
"access": "Transition Projects (TPI) - ACCESS - CM(5471)",
"res": "Transition Projects (TPI) - Residential - CM(5473)",
"ret": "Transition Projects (TPI) - Retention - CM(5472)",
"ca": "Transition Projects (TPI) Housing - Clark Annex PSH - SP(2858)",
"cagpd": "Transition Projects (TPI) Housing - Clark Annex GPD - SP(4259)"
}
provider_list = []
for dept in provider:
name = full_provider_name[dept.lower()]
provider_list.append(name)
count = len(entries_df[
entries_df["Entry Exit Provider Id"].isin(provider_list)
].drop_duplicates(subset="Client Uid", keep="first").index)
return tuple([
"1850 unduplicated participants in all emergency shelters will have a safe place to sleep",
1850,
count
])
def count_exit_destination_by_shelter_group(self, entries_df, shelter_group):
low_barrier = [
"Transition Projects (TPI) - SOS Shelter(2712)",
"Transition Projects (TPI) - Willamette Center(5764)",
"Transition Projects (TPI) - Columbia Shelter(5857)",
"Transition Projects (TPI) - 5th Avenue Shelter(6281)",
"Transition Projects (TPI) - Hansen Emergency Shelter - SP(5588)",
]
residential = [
"Transition Projects(TPI) - Clark Center - SP(25)",
"Transition Projects(TPI) - VA Grant Per Diem(inc.Doreen's Place GPD) - SP(3189)",
"Transition Projects(TPI) - Jean's Place L1 - SP(29)",
"Transition Projects(TPI) - Doreen's Place - SP(28)",
"Transition Projects(TPI) - Jean’s Place VA Grant Per Diem(GPD) - SP(3362)"
]
perm_destination = [
"Owned by client, no ongoing housing subsidy (HUD)",
"Owned by client, with ongoing housing subsidy (HUD)",
"Permanent housing for formerly homeless persons (HUD)",
"Rental by client, no ongoing housing subsidy (HUD)",
"Rental by client, with other ongoing housing subsidy (HUD)",
"Rental by client, with VASH subsidy (HUD)",
"Staying or living with family, permanent tenure (HUD)",
"Staying or living with friends, permanent tenure (HUD)",
"Foster care home or foster care group home (HUD)",
"Rental by client, with GPD TIP subsidy (HUD)",
"Permanent housing (other than RRH) for formerly homeless persons (HUD)",
"Moved from one HOPWA funded project to HOPWA PH (HUD)",
"Long-term care facility or nursing home (HUD)",
"Residential project or halfway house with no homeless criteria (HUD)"
]
temp_destination = [
# "Emergency shelter, including hotel or motel paid for with emergency shelter voucher (HUD)",
"Hospital or other residential non-psychiatric medical facility (HUD)",
"Hotel or motel paid for without emergency shelter voucher (HUD)",
"Jail, prison or juvenile detention facility (HUD)",
"Staying or living with family, temporary tenure (e.g., room, apartment or house)(HUD)",
"Staying or living with friends, temporary tenure (e.g., room apartment or house)(HUD)",
"Transitional housing for homeless persons (including homeless youth) (HUD)",
"Moved from one HOPWA funded project to HOPWA TH (HUD)",
"Substance abuse treatment facility or detox center (HUD)",
"Psychiatric hospital or other psychiatric facility (HUD)"
]
if shelter_group.lower() == "res":
perm = entries_df[
entries_df["Entry Exit Provider Id"].isin(residential) &
entries_df["Entry Exit Destination"].notna() &
entries_df["Entry Exit Destination"].isin(perm_destination)
].drop_duplicates(subset="Client Uid")
temp = entries_df[
entries_df["Entry Exit Provider Id"].isin(residential) &
entries_df["Entry Exit Destination"].notna() &
entries_df["Entry Exit Destination"].isin(temp_destination)
].drop_duplicates(subset="Client Uid")
all = entries_df[
entries_df["Entry Exit Provider Id"].isin(residential) &
entries_df["Entry Exit Exit Date"].notna()
].drop_duplicates(subset="Client Uid")
return (len(perm.index), len(temp.index), len(all.index))
elif shelter_group.lower() == "low":
perm = entries_df[
entries_df["Entry Exit Provider Id"].isin(low_barrier) &
entries_df["Entry Exit Destination"].notna() &
entries_df["Entry Exit Destination"].isin(perm_destination)
].drop_duplicates(subset="Client Uid")
temp = entries_df[
entries_df["Entry Exit Provider Id"].isin(low_barrier) &
entries_df["Entry Exit Destination"].notna() &
entries_df["Entry Exit Destination"].isin(temp_destination)
].drop_duplicates(subset="Client Uid")
all = entries_df[
entries_df["Entry Exit Provider Id"].isin(low_barrier) &
entries_df["Entry Exit Exit Date"].notna()
].drop_duplicates(subset="Client Uid")
return (len(perm.index), len(temp.index), len(all.index))
else:
pass
def count_households_screened(self, entries_hh_df):
"""
Used By:
:param entries_hh_df:
:return:
"""
screened = len(entries_hh_df[
entries_hh_df["Entry Exit Provider Id"].str.contains("Screening")
].drop_duplicates(subset="Household Uid", keep="first").index)
return tuple(["Screen 784 veteran families for services", 784, screened])
def count_hygiene_services_by_provider(self, services_df, provider="Day Center"):
"""
Use for: Agency, Day Center
Question: participants will receive hygiene services
Warning: Do not update the services_1 list to include the newer service code description of
"Personal Goods/Services" as this will cause the module to return a count including non-hygiene services that
are sharing this same service code description
:return: a count of unique participants receiving any hygiene service
"""
services_1 = [
"Bathing Facilities",
"Personal/Grooming Supplies",
"Hairdressing/Nail Care"
]
services_2 = [
"Shower",
"Showers",
"Laundry Supplies",
"Clothing",
"Hairdressing/Nail Care",
"Personal Grooming Supplies"
]
if provider == "Day Center":
services = services_df[
((services_df["Service Code Description"].isin(services_1)) | (
services_df["Service Provider Specific Code"].isin(services_2))) &
services_df["Service Provide Provider"].str.contains(provider)
]
services_provided = services[
services["Service Provide Provider"].str.contains("Day Center")
]
return tuple(["40,000 hygiene services provided", 40000, len(services_provided.index)])
elif provider == "Agency":
services = services_df[
((services_df["Service Code Description"].isin(services_1)) | (
services_df["Service Provider Specific Code"].isin(services_2)))
]
de_duped = services.drop_duplicates(subset="Client Uid", inplace=False)
return tuple(["7,500 participants will receive hygiene services", 7500, len(de_duped.index)])
def count_id_assistance_by_provider(self, services_df, provider="Day Center"):
"""
Used by: Day Center
:param services_df:
:param provider:
:return:
"""
id_assistance = ["Birth Certificate", "Driver's License/State ID Card"]
served = services_df[
services_df["Service Provide Provider"].str.contains(provider) & services_df[
"Service Provider Specific Code"].isin(id_assistance)
].drop_duplicates(subset="Client Uid")
return tuple(["1500 individuals received assistance obtaining ID documents", 1500, len(served.index)])
def count_mailing_services_by_day_center(self, services_df):
"""
Used For: Day Center
:param services_df:
:return:
"""
mail_services = len(services_df[
services_df["Service Code Description"] == "Temporary Mailing Address"
].index)
return tuple(["43,000 mailing services provided", 43000, mail_services])
def count_ongoing_cm_services(self, services_df):
"""
Used For: Agency
Issues: Currently, regardless of the method name, this method counts unique individuals served with a CM service
ignoring department. This could cause false counts when the wrong service is selected by a non-cm provider and
does not truly show ongoing services as defined by the agency data dictionary.
:param services_df:
:return:
"""
cm_services = [
"Case Management - Office Visit",
"Case Management - Other",
"Case Management - Phone Meeting",
"Case Management - Home Visit",
"Case Management Meeting - Home Visit",
"Case Management Meeting - Office Visit",
"Case Management Meeting - Phone"
]
receiving = len(services_df[
services_df["Service Provider Specific Code"].isin(cm_services)
].drop_duplicates(subset="Client Uid", inplace=False).index)
return tuple(["2,100 participants served through case management", 2100, receiving])
def count_ongoing_cm_services_by_department(self, services_df, provider="SSVF"):
"""
Used For: Residential CM, Retention, SSVF
Question: Provide ongoing case management to X participants
:param services_df:
:param provider:
:return:
"""
cm_services = [
"Case Management - Office Visit",
"Case Management - Other",
"Case Management - Phone Meeting",
"Case Management - Home Visit",
"Case Management Meeting - Home Visit",
"Case Management Meeting - Office Visit",
"Case Management Meeting - Phone"
]
in_provider = services_df[
services_df["Service Provide Provider"].str.contains(provider)
]
receiving = in_provider[
in_provider["Service Provider Specific Code"].isin(cm_services)
]
ongoing = receiving.groupby(by="Client Uid").count()
output = len(ongoing[ongoing["Service Provider Specific Code"] >= 2].index)
if provider == "Residential":
return tuple(["Provide ongoing case management to 700 participants", 700, output])
elif provider == "Retention":
return tuple(["Provide ongoing case management to 800 participants", 800, output])
elif provider == "SSVF":
return tuple(["Provide ongoing case management to 450 participants", 450, output])
elif provider == "ACCESS":
return tuple(["Provide ongoing case management to 800 participants", 800, output])
else:
return tuple(["PROVIDERERROR", "PROVIDERERROR", "PROVIDERERROR"])
def count_perm_by_provider(self, placements_df, provider=["ACCESS", "SSVF - TPI", "Retention", "Residential CM"]):
"""
Used For: Outreach, SSVF, Retention CM, Residential CM
:param placements_df:
:param provider:
:return:
"""
placements = len(placements_df[
(placements_df["Department Placed From(3076)"].isin(provider)) & (
placements_df["Intervention Type (TPI)(8745)"] == "Permanent Placement")
].drop(
[
"Client First Name",
"Client Last Name",
"Department Placed From(3076)",
"Placement Case Manager(3075)",
"Placement Grant(8743)",
"Reporting Program (TPI)(8748)"
],
axis=1
).index)
if (len(provider) == 1) and (provider[0] == "ACCESS"):
return tuple(["415 participants move into permanent housing", 415, placements])
elif (len(provider) == 1) and (provider[0] == "SSVF - TPI"):
return tuple(["262 veteran families will move into permanent housing", 262, placements])
else:
return "Error: Empty Provider List"
def count_pts_with_barrier_mitigation_and_doc_prep(self, services_df, provider="CHAT"):
"""
Used by: Coordinated Access (CHAT)
:param services_df: Use the standard services spread sheet.
:param provider:
:return:
"""
services = [
"Housing Barrier Resolution",
"Birth Certificate",
"DD214",
"Driver's License/State ID Card",
"General Form Assistance",
"Notary Service"
]
served = len(services_df[
services_df["Service Provide Provider"].str.contains(provider) &
services_df["Service Provider Specific Code"].isin(services)
].drop_duplicates(subset="Client Uid", inplace=False).index)
return tuple([
"Provide barrier mitigation and document prep to 150 individuals",
150,
served
])
def count_ep_by_provider(self, placements_df, provider=["ACCESS", "SSVF - TPI", "Retention", "Residential CM"]):
"""
Used For: Outreach, SSVF, Retention CM, Residential CM
:param placements_df:
:param provider:
:return:
"""
placements = placements_df[
(placements_df["Department Placed From(3076)"].isin(provider)) & (
placements_df["Intervention Type (TPI)(8745)"] == "Eviction Prevention")
].drop(
[
"Client First Name",
"Client Last Name",
"Department Placed From(3076)",
"Placement Case Manager(3075)",
"Placement Grant(8743)",
"Reporting Program (TPI)(8748)"
],
axis=1
)
if (len(provider) == 1) and (provider[0] == "SSVF - TPI"):
return tuple(["56 veteran families will have evictions prevented", 56, len(placements.index)])
else:
return tuple(["", "", len(placements.index)])
def count_exclusions_by_provider(self, exclusions_df, provider):
exclusions = exclusions_df[
exclusions_df["Infraction Provider"].str.contains(provider) &
exclusions_df["Infraction Banned Code"].notna() &
(exclusions_df["Infraction Banned Code"] != "Warning") &
(exclusions_df["Infraction Banned Code"] != "Safety Alert") &
(exclusions_df["Infraction Banned Code"] != "Other")
]
return tuple([
"20% reduction in participant exclusions",
"20%",
"{} <--- Must be divided by last quarter's numbers -(last quarter/ this quarter)".format(
len(exclusions.index)
)
])
def count_latinos_served_by_provider(self, services_df, provider="Wellness Access"):
"""
Used For: Wellness Access
Use: Standard All Services Report
:param services_df:
:param provider:
:return:
"""
original = services_df
cleaned = original[
(
original["Service Provide Provider"].str.contains(provider) &
(original["Ethnicity (Hispanic/Latino)(896)"].str.contains("Hispanic/Latino"))
)
].drop_duplicates(subset="Client Uid")
return tuple(["80 Latino participants outreached to per year", 80, len(cleaned.index)])
def count_legal_barriers_mitigated(self, entries_df, services_df, provider):
"""
Used For: SSVF
Needs to be modified to look at the version of the all services report which includes needs outcomes.
:param cm_provider:
:param services_df:
:param entries_df:
:return: a count of the participants
"""
no_na = entries_df.dropna(axis=0, subset=["Entry Exit Provider Id"])
entries = no_na[no_na["Entry Exit Provider Id"].str.contains(provider)]
in_provider_list = entries["Client Uid"].tolist()
legal_services = len(services_df[
(
(services_df["Service Code Description"] == "Legal Services") &
(services_df["Client Uid"].isin(in_provider_list))
)
].drop_duplicates(subset="Client Uid").index)
return tuple(["50 veteran families will have legal barriers mitigated", 50, legal_services])
def count_poc_placed(self, placements_df, services_df):
"""
Used For: Agency
:param placements_df:
:return:
"""
poc_placements = len(placements_df[
placements_df["Client Uid"].isin(self.return_poc_list(services_df))
].drop_duplicates(subset="Client Uid").index)
return poc_placements
def count_poc_placed_by_provider(
self,
placements_df,
services_df,
provider=["ACCESS", "SSVF - TPI", "Retention", "Residential CM"]
):
"""
Used For:
:param placements_df:
:param services_df:
:param provider:
:return:
"""
poc_placements = placements_df[
(
(placements_df["Client Uid"].isin(self.return_poc_list(services_df))) &
(placements_df["Department Placed From(3076)"].isin(provider))
)
]
return len(poc_placements.drop_duplicates(subset="Client Uid").index)
def count_provider(self, entries_df, cm_provider, goal):
"""
Used For: Agency
:param entries_df:
:param cm_provider:
:param goal:
:return:
"""
provider_ee = len(entries_df[
entries_df["Entry Exit Provider Id"].str.contains(cm_provider)
].drop_duplicates(subset="Client Uid", keep="first").index)
return tuple(["{} participants served by {}".format(goal, cm_provider), goal, provider_ee])
def count_referrals_resulting_in_connections(self, services_df, needs_df, provider, referrals, metric):
"""
Used by: Wellness Access
Method of the method: First remove rows from the services data frame which were not created by the provider and
were not in the list of referral services.
Then, remove rows from the needs data frame where the Client Uid is not in the Client Uid column of the services
data frame and the need status is not fully met.
Merge left the needs data frame into the services data frame using Client Uid from both data frames as well as
Need Creation Date from the need data frame (right) and the service creation data from the services data frame
(left). This will be done using the pd.merge method and entering the columns as lists of strings in the
left_on and right_on params.
medical_referrals = ["Referral - Eye Care", "Referral - Dental Care", "Referral - Medical Care"]
mh_referrals = [
"Referral - A&D Support", "Referral - DV Support", "Referral - Mental Health Care", "Referral - MH Support"
]
:param provider:
:param services_df:
:param needs_df:
:param referrals: provide a list of strings or this will return an error
:param metric: enter one of the following strings - 'med count', 'mh sud count', 'percent med', 'percent mh sud'
:return:
"""
services = services_df[
services_df["Service Provide Provider"].str.contains(provider) &
services_df["Service Provider Specific Code"].isin(referrals)
]
needs = needs_df[
needs_df["Client Uid"].isin(services["Client Uid"].tolist()) &
(needs_df["Need Status"] == "Closed") &
(needs_df["Need Outcome"] == "Fully Met")
]
served = pd.merge(
services,
needs,
how="left",
left_on=["Client Uid", "Service Provide Start Date"],
right_on=["Client Uid", "Need Date Set"]
)
if metric == "med count":
return tuple([
"200 connections to medical care per year",
200,
len(served.index)
])
elif metric == "mh sud count":
return tuple([
"700 connections to mental health or SUD services per year",
700,
len(served.index)
])
elif metric == "percent med":
all = len(served.index)
clean = served.dropna(axis=0, how="any", subset=["Need Uid"])
success = len(clean.index)
return tuple([
"50% of referrals result in connection to medical care provider",
"50%",
"{}/{} = {}%".format(success, all, 100*(success/all))
])
elif metric == "percent mh sud":
all = len(served.index)
success = len(served.dropna(axis=0, how="any", subset=["Need Uid"]).index)
return tuple([
"50% of referrals result in connection to mental health and/or SUD services",
"50%",
"{}/{} = {}%".format(success, all, 100 * (success / all))
])
else:
return "Param Error: metric's value was not among the list of used values"
def count_rent_assist(self, services_df):
"""
Use by: Agency
Question: participants will receive rent assistance
:return: a count of unique participants receiving any rent assistance service
"""
rent_services = [
"Rent Payment Assistance",
"Rental Application Fee Payment Assistance",
"Rental Deposit Assistance"
]
rent_service_2 = [
"Application Fee",
"Arrears / Property Debt",
"Deposit",
"Rent Payment Assistance"
]
services = services_df[
(services_df["Service Code Description"].isin(rent_services)) | (
services_df["Service Provider Specific Code"].isin(rent_service_2))
]
unique = len(services.drop_duplicates(
subset="Client Uid",
keep="first",
inplace=False
).index)
return tuple(["900 participants will receive rent assistance", 900, unique])
def count_rent_well(self, services_df, type):
"""
Used by: Agency, RentWell
This method will identify the participants who either had at least a
single service by TPI RentWell or a graduation service during the
reporting period.
:param services_df:
:return:
"""
if type.lower() == "attendance":
attended = services_df[
services_df["Service Provider Specific Code"].notna() &
(
services_df["Service Provider Specific Code"].str.contains("RentWell - Attendence") |
services_df["Service Provider Specific Code"].str.contains("RentWell - Graduation")
)
].drop_duplicates(subset="Client Uid")
return tuple(["400 participants will enroll in Rent Well ", 400, len(attended.index)])
elif type.lower() == "graduation":
graduated = services_df[
services_df["Service Provider Specific Code"].notna() &
services_df["Service Provider Specific Code"].str.contains("RentWell - Graduation")
].drop_duplicates(subset="Client Uid")
return tuple(["240 Participants will graduate RentWell", 240, len(graduated.index)])
elif type.lower() == "services":
services = services_df[
services_df["Service Provider Specific Code"].notna() &
(
services_df["Service Provider Specific Code"].str.contains("RentWell - Attendence") |
services_df["Service Provider Specific Code"].str.contains("RentWell - Graduation")
)
]
return tuple(["", "", len(services.index)])
elif type.lower() == "all served":
attended = services_df[
services_df["Service Provider Specific Code"].notna() &
(
services_df["Service Provider Specific Code"].str.contains("RentWell - Attendence") |
services_df["Service Provider Specific Code"].str.contains("RentWell - Graduation")
)
].drop_duplicates(subset="Client Uid")
return attend[["Client Uid", "Service Provide Start Date"]]
elif type.lower() == "all graduates":
graduated = services_df[
services_df["Service Provider Specific Code"].notna() &
services_df["Service Provider Specific Code"].str.contains("RentWell - Graduation")
].drop_duplicates(subset="Client Uid")
return graduated[["Client Uid", "Service Provide Start Date"]]
else:
return tuple(["ERROR", "ERROR", "ERROR"])
def count_retention_by_length(self, retention_df, length, provider="agency"):
"""
Used by: Agency
For this to work you need to add a Months Post Subsidy column to the follow ups report. The column must be filled down with the following formula:
=IF(ISBLANK(B2),"",DATEDIF(B2,C2,"M"))
:param length: int: 1-12
:return:
"""
no_na = retention_df.dropna(axis=0, subset=["Months Post Subsidy"])
fu_of_duration = no_na[
(
(no_na["Months Post Subsidy"] >= length - 1) &
(no_na["Months Post Subsidy"] <= length + 1)
)